






"""simple docstring"""







import warnings
from typing import List, Optional, Union

import numpy as np
import PIL
import torch
from PIL import Image

from .configuration_utils import ConfigMixin, register_to_config
from .utils import CONFIG_NAME, PIL_INTERPOLATION, deprecate







class 							_SCREAMING_SNAKE_CASE    (	__snake_case				):





							"""simple docstring"""





							_SCREAMING_SNAKE_CASE     =CONFIG_NAME




							@register_to_config
							def __init__( self:  int					,		__A:  bool = True					,		__A:  int = 8					,		__A:  str = "lanczos"					,		__A:  bool = True					,		__A:  bool = False					,		):

												'''simple docstring'''

												super().__init__()




							@staticmethod
							def 			lowercase		( __A:  np.ndarray     ):

												'''simple docstring'''

												if images.ndim == 3:
																	a__ 							=	images[None, ...]
												a__ 							=	(images * 255).round().astype('''uint8'''     )
												if images.shape[-1] == 1:
																	# special case for grayscale (single channel) images
																	a__ 							=	[Image.fromarray(image.squeeze()					,		mode='''L'''     ) for image in images]
												else:
																	a__ 							=	[Image.fromarray(__A     ) for image in images]

												return pil_images




							@staticmethod
							def 			lowercase		( __A:  Union[List[PIL.Image.Image], PIL.Image.Image]     ):

												'''simple docstring'''

												if not isinstance(__A					,		__A     ):
																	a__ 							=	[images]
												a__ 							=	[np.array(__A     ).astype(np.floataa     ) / 2_5_5.0 for image in images]
												a__ 							=	np.stack(__A					,		axis=0     )

												return images




							@staticmethod
							def 			lowercase		( __A:  np.ndarray     ):

												'''simple docstring'''

												if images.ndim == 3:
																	a__ 							=	images[..., None]

												a__ 							=	torch.from_numpy(images.transpose(0					,		3					,		1					,		2     )     )
												return images




							@staticmethod
							def 			lowercase		( __A:  torch.FloatTensor     ):

												'''simple docstring'''

												a__ 							=	images.cpu().permute(0					,		2					,		3					,		1     ).float().numpy()
												return images




							@staticmethod
							def 			lowercase		( __A:  int     ):

												'''simple docstring'''

												return 2.0 * images - 1.0




							@staticmethod
							def 			lowercase		( __A:  Any     ):

												'''simple docstring'''

												return (images / 2 + 0.5).clamp(0					,		1     )




							@staticmethod
							def 			lowercase		( __A:  PIL.Image.Image     ):

												'''simple docstring'''

												a__ 							=	image.convert('''RGB'''     )
												return image




							def 			lowercase		( self:  List[str]					,		__A:  PIL.Image.Image					,		__A:  Optional[int] = None					,		__A:  Optional[int] = None					,		):

												'''simple docstring'''

												if height is None:
																	a__ 							=	image.height
												if width is None:
																	a__ 							=	image.width

												a__							,a__ 							=	(
												    x - x % self.config.vae_scale_factor for x in (width, height)
												)  # resize to integer multiple of vae_scale_factor
												a__ 							=	image.resize((width, height)					,		resample=PIL_INTERPOLATION[self.config.resample]     )
												return image




							def 			lowercase		( self:  int					,		__A:  Union[torch.FloatTensor, PIL.Image.Image, np.ndarray]					,		__A:  Optional[int] = None					,		__A:  Optional[int] = None					,		):

												'''simple docstring'''

												a__ 							=	(PIL.Image.Image, np.ndarray, torch.Tensor)
												if isinstance(__A					,		__A     ):
																	a__ 							=	[image]
												elif not (isinstance(__A					,		__A     ) and all(isinstance(__A					,		__A     ) for i in image     )):
																	raise ValueError(
																	    F'Input is in incorrect format: {[type(__A     ) for i in image]}. Currently, we only support {", ".join(__A     )}'     )

												if isinstance(image[0]					,		PIL.Image.Image     ):
																	if self.config.do_convert_rgb:
																						a__ 							=	[self.convert_to_rgb(__A     ) for i in image]
																	if self.config.do_resize:
																						a__ 							=	[self.resize(__A					,		__A					,		__A     ) for i in image]
																	a__ 							=	self.pil_to_numpy(__A     )  # to np
																	a__ 							=	self.numpy_to_pt(__A     )  # to pt

												elif isinstance(image[0]					,		np.ndarray     ):
																	a__ 							=	np.concatenate(__A					,		axis=0     ) if image[0].ndim == 4 else np.stack(__A					,		axis=0     )
																	a__ 							=	self.numpy_to_pt(__A     )
																	a__							,a__							,a__							,a__ 							=	image.shape
																	if self.config.do_resize and (
																	    height % self.config.vae_scale_factor != 0 or width % self.config.vae_scale_factor != 0
																	):
																						raise ValueError(
																						    F'Currently we only support resizing for PIL image - please resize your numpy array to be divisible by {self.config.vae_scale_factor}'
																						    F'currently the sizes are {height} and {width}. You can also pass a PIL image instead to use resize option in VAEImageProcessor'     )

												elif isinstance(image[0]					,		torch.Tensor     ):
																	a__ 							=	torch.cat(__A					,		axis=0     ) if image[0].ndim == 4 else torch.stack(__A					,		axis=0     )
																	a__							,a__							,a__							,a__ 							=	image.shape

																	# don't need any preprocess if the image is latents
																	if channel == 4:
																						return image

																	if self.config.do_resize and (
																	    height % self.config.vae_scale_factor != 0 or width % self.config.vae_scale_factor != 0
																	):
																						raise ValueError(
																						    F'Currently we only support resizing for PIL image - please resize your pytorch tensor to be divisible by {self.config.vae_scale_factor}'
																						    F'currently the sizes are {height} and {width}. You can also pass a PIL image instead to use resize option in VAEImageProcessor'     )

        # expected range [0,1], normalize to [-1,1]
												a__ 							=	self.config.do_normalize
												if image.min() < 0:
																	warnings.warn(
																	    '''Passing `image` as torch tensor with value range in [-1,1] is deprecated. The expected value range for image tensor is [0,1] '''
																	    F'when passing as pytorch tensor or numpy Array. You passed `image` with value range [{image.min()},{image.max()}]'					,		__A					,		)
																	a__ 							=	False

												if do_normalize:
																	a__ 							=	self.normalize(__A     )

												return image





							def 			lowercase		( self:  str					,		__A:  torch.FloatTensor					,		__A:  str = "pil"					,		__A:  Optional[List[bool]] = None					,		):

												'''simple docstring'''

												if not isinstance(__A					,		torch.Tensor     ):
																	raise ValueError(
																	    F'Input for postprocessing is in incorrect format: {type(__A     )}. We only support pytorch tensor'     )
												if output_type not in ["latent", "pt", "np", "pil"]:
																	a__ 							=	(
																	    F'the output_type {output_type} is outdated and has been set to `np`. Please make sure to set it to one of these instead: '
																	    '''`pil`, `np`, `pt`, `latent`'''
																	)
																	deprecate('''Unsupported output_type'''					,		'''1.0.0'''					,		__A					,		standard_warn=__A     )
																	a__ 							=	'''np'''

												if output_type == "latent":
																	return image

												if do_denormalize is None:
																	a__ 							=	[self.config.do_normalize] * image.shape[0]

												a__ 							=	torch.stack(
												    [self.denormalize(image[i]     ) if do_denormalize[i] else image[i] for i in range(image.shape[0]     )]     )

												if output_type == "pt":
																	return image

												a__ 							=	self.pt_to_numpy(__A     )

												if output_type == "np":
																	return image

												if output_type == "pil":
																	return self.numpy_to_pil(__A     )







class 							_SCREAMING_SNAKE_CASE    (	__snake_case				):





							"""simple docstring"""





							_SCREAMING_SNAKE_CASE     =CONFIG_NAME




							@register_to_config
							def __init__( self:  List[Any]					,		__A:  bool = True					,		__A:  int = 8					,		__A:  str = "lanczos"					,		__A:  bool = True					,		):

												'''simple docstring'''

												super().__init__()




							@staticmethod
							def 			lowercase		( __A:  Optional[int]     ):

												'''simple docstring'''

												if images.ndim == 3:
																	a__ 							=	images[None, ...]
												a__ 							=	(images * 255).round().astype('''uint8'''     )
												if images.shape[-1] == 1:
																	# special case for grayscale (single channel) images
																	a__ 							=	[Image.fromarray(image.squeeze()					,		mode='''L'''     ) for image in images]
												else:
																	a__ 							=	[Image.fromarray(image[:, :, :3]     ) for image in images]

												return pil_images




							@staticmethod
							def 			lowercase		( __A:  str     ):

												'''simple docstring'''

												return image[:, :, 1] * 2**8 + image[:, :, 2]




							def 			lowercase		( self:  List[Any]					,		__A:  Union[str, Any]     ):

												'''simple docstring'''

												if images.ndim == 3:
																	a__ 							=	images[None, ...]
												a__ 							=	images[:, :, :, 3:]
												if images.shape[-1] == 6:
																	a__ 							=	(images_depth * 255).round().astype('''uint8'''     )
																	a__ 							=	[
																	    Image.fromarray(self.rgblike_to_depthmap(__A     )					,		mode='''I;16'''     ) for image_depth in images_depth
																	]
												elif images.shape[-1] == 4:
																	a__ 							=	(images_depth * 6_5_5_3_5.0).astype(np.uintaa     )
																	a__ 							=	[Image.fromarray(__A					,		mode='''I;16'''     ) for image_depth in images_depth]
												else:
																	raise Exception('''Not supported'''     )

												return pil_images





							def 			lowercase		( self:  List[Any]					,		__A:  torch.FloatTensor					,		__A:  str = "pil"					,		__A:  Optional[List[bool]] = None					,		):

												'''simple docstring'''

												if not isinstance(__A					,		torch.Tensor     ):
																	raise ValueError(
																	    F'Input for postprocessing is in incorrect format: {type(__A     )}. We only support pytorch tensor'     )
												if output_type not in ["latent", "pt", "np", "pil"]:
																	a__ 							=	(
																	    F'the output_type {output_type} is outdated and has been set to `np`. Please make sure to set it to one of these instead: '
																	    '''`pil`, `np`, `pt`, `latent`'''
																	)
																	deprecate('''Unsupported output_type'''					,		'''1.0.0'''					,		__A					,		standard_warn=__A     )
																	a__ 							=	'''np'''

												if do_denormalize is None:
																	a__ 							=	[self.config.do_normalize] * image.shape[0]

												a__ 							=	torch.stack(
												    [self.denormalize(image[i]     ) if do_denormalize[i] else image[i] for i in range(image.shape[0]     )]     )

												a__ 							=	self.pt_to_numpy(__A     )

												if output_type == "np":
																	if image.shape[-1] == 6:
																						a__ 							=	np.stack([self.rgblike_to_depthmap(im[:, :, 3:]     ) for im in image]					,		axis=0     )
																	else:
																						a__ 							=	image[:, :, :, 3:]
																	return image[:, :, :, :3], image_depth

												if output_type == "pil":
																	return self.numpy_to_pil(__A     ), self.numpy_to_depth(__A     )
												else:
																	raise Exception(F'This type {output_type} is not supported'     )










"""simple docstring"""







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

from transformers import BigBirdTokenizerFast


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





def 							SCREAMING_SNAKE_CASE							(   lowerCamelCase_			,  lowerCamelCase_=0			,  lowerCamelCase_=None):
					return [" ".join(x.split(''' ''')[begin:end]) for x in answers if len(x.split(''' ''')) > 1]





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





def 							SCREAMING_SNAKE_CASE							(   lowerCamelCase_			,  lowerCamelCase_			,  lowerCamelCase_=1			,  lowerCamelCase_=100):
					a__							,a__ 							=	jax.lax.top_k(lowerCamelCase_			,  lowerCamelCase_)
					a__							,a__ 							=	jax.lax.top_k(lowerCamelCase_			,  lowerCamelCase_)

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

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





def 							SCREAMING_SNAKE_CASE							(   lowerCamelCase_):
					a__ 							=	sample['''question''']['''text''']
					a__ 							=	sample['''document''']['''tokens''']['''token''']
					a__ 							=	sample['''document''']['''tokens''']['''is_html''']
					a__ 							=	sample['''annotations''']['''long_answer''']
					a__ 							=	sample['''annotations''']['''short_answers''']

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

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

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

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

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

					a__ 							=	'''vasudevgupta/flax-bigbird-natural-questions'''
					a__ 							=	FlaxBigBirdForNaturalQuestions.from_pretrained(lowerCamelCase_)
					a__ 							=	BigBirdTokenizerFast.from_pretrained(lowerCamelCase_)

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

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

										a__							,a__							,a__ 							=	forward(**lowerCamelCase_)

										a__ 							=	CATEGORY_MAPPING[category.item()]

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

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

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

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

										a__ 							=	expand_to_aliases(example['''targets''']			,  make_sub_answers=lowerCamelCase_)
										a__ 							=	expand_to_aliases(example['''output'''])

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

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

										return example

					a__ 							=	short_validation_dataset.map(lowerCamelCase_)

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


if __name__ == "__main__":
	main()


