


import logging
import math
from functools import partial
from typing import Any, Callable, Dict, Iterable, List, Optional, Sequence, Tuple, Union

import torch

from .tensor_utils import tensor_tree_map, tree_map


def 				lowerCamelCase__	(      a	:   Tuple				)    ->	Union[str, Any]:





			"""simple docstring"""
			a__ :List[Any]						  =    []
			if isinstance(__lowerCAmelCase	,  __lowerCAmelCase				):
						for v in tree.values():
									shapes.extend(_fetch_dims(__lowerCAmelCase				)				)
			elif isinstance(__lowerCAmelCase	,  (list, tuple)				):
						for t in tree:
									shapes.extend(_fetch_dims(__lowerCAmelCase				)				)
			elif isinstance(__lowerCAmelCase	,  torch.Tensor				):
						shapes.append(tree.shape				)
			else:
						raise ValueError("Not supported"				)

			return shapes


@torch.jit.ignore
def 				lowerCamelCase__	(      a	:   int	,  a	:   List[str]				)    ->	List[str]:





			"""simple docstring"""
			a__ :List[Any]						  =    []
			for d in reversed(__lowerCAmelCase				):
						idx.append(flat_idx % d				)
						a__ :Dict						  =    flat_idx // d

			return tuple(reversed(__lowerCAmelCase				)				)


@torch.jit.ignore
def 				lowerCamelCase__	(      a	:   Any	,  a	:   List[str]	,  a	:   List[str]	,  a	:   str = None	,  a	:   Tuple = None	,  )    ->	Any:





			"""simple docstring"""

			# start_edges and end_edges both indicate whether, starting from any given
			# dimension, the start/end index is at the top/bottom edge of the
			# corresponding tensor, modeled as a tree
			def reduce_edge_list(a	:   Optional[int]				) -> None:
						a__ :List[Any]						  =    True
						for i in range(len(__lowerCAmelCase				)				):
									a__ :Tuple						  =    -1 * (i + 1)
									l[reversed_idx] &= tally
									a__ :str						  =    l[reversed_idx]

			if start_edges is None:
						a__ :Union[str, Any]						  =    [s == 0 for s in start]
						reduce_edge_list(__lowerCAmelCase				)
			if end_edges is None:
						a__ :List[str]						  =    [e == (d - 1) for e, d in zip(__lowerCAmelCase	,  __lowerCAmelCase				)]
						reduce_edge_list(__lowerCAmelCase				)

			# Base cases. Either start/end are empty and we're done, or the final,
			# one-dimensional tensor can be simply sliced
			if len(__lowerCAmelCase				) == 0:
						return [()]
			elif len(__lowerCAmelCase				) == 1:
						return [(slice(start[0]	,  end[0] + 1				),)]

			a__ :List[Tuple[slice, ...]]						  =    []
			a__ :List[slice]						  =    []

			# Dimensions common to start and end can be selected directly
			for s, e in zip(__lowerCAmelCase	,  __lowerCAmelCase				):
						if s == e:
									path_list.append(slice(__lowerCAmelCase	,  s + 1				)				)
						else:
									break

			a__ :Tuple[slice, ...]						  =    tuple(__lowerCAmelCase				)
			a__ :Tuple						  =    len(__lowerCAmelCase				)

			# start == end, and we're done
			if divergence_idx == len(__lowerCAmelCase				):
						return [path]

			def upper() -> Tuple[Tuple[slice, ...], ...]:
						assert start_edges is not None
						assert end_edges is not None

						a__ :List[str]						  =    start[divergence_idx]
						return tuple(
						    path + (slice(__lowerCAmelCase	,  sdi + 1				),) + s
						    for s in _get_minimal_slice_set(
						        start[divergence_idx + 1 :]	,  [d - 1 for d in dims[divergence_idx + 1 :]]	,  dims[divergence_idx + 1 :]	,  start_edges=start_edges[divergence_idx + 1 :]	,  end_edges=[True for _ in end_edges[divergence_idx + 1 :]]	,  )				)

			def lower() -> Tuple[Tuple[slice, ...], ...]:
						assert start_edges is not None
						assert end_edges is not None

						a__ :int						  =    end[divergence_idx]
						return tuple(
						    path + (slice(__lowerCAmelCase	,  edi + 1				),) + s
						    for s in _get_minimal_slice_set(
						        [0 for _ in start[divergence_idx + 1 :]]	,  end[divergence_idx + 1 :]	,  dims[divergence_idx + 1 :]	,  start_edges=[True for _ in start_edges[divergence_idx + 1 :]]	,  end_edges=end_edges[divergence_idx + 1 :]	,  )				)

			# If both start and end are at the edges of the subtree rooted at
			# divergence_idx, we can just select the whole subtree at once
			if start_edges[divergence_idx] and end_edges[divergence_idx]:
						slices.append(path + (slice(start[divergence_idx]	,  end[divergence_idx] + 1				),)				)
			# If just start is at the edge, we can grab almost all of the subtree,
			# treating only the ragged bottom edge as an edge case
			elif start_edges[divergence_idx]:
						slices.append(path + (slice(start[divergence_idx]	,  end[divergence_idx]				),)				)
						slices.extend(lower()				)
			# Analogous to the previous case, but the top is ragged this time
			elif end_edges[divergence_idx]:
						slices.extend(upper()				)
						slices.append(path + (slice(start[divergence_idx] + 1	,  end[divergence_idx] + 1				),)				)
			# If both sides of the range are ragged, we need to handle both sides
			# separately. If there's contiguous meat in between them, we can index it
			# in one big chunk
			else:
						slices.extend(upper()				)
						a__ :str						  =    end[divergence_idx] - start[divergence_idx]
						if middle_ground > 1:
									slices.append(path + (slice(start[divergence_idx] + 1	,  end[divergence_idx]				),)				)
						slices.extend(lower()				)

			return slices


@torch.jit.ignore
def 				lowerCamelCase__	(      a	:   Dict	,  a	:   Union[str, Any]	,  a	:   List[str]	,  a	:   Any				)    ->	Union[str, Any]:





			"""simple docstring"""
			a__ :Any						  =    t.shape[:no_batch_dims]
			a__ :Dict						  =    list(_flat_idx_to_idx(__lowerCAmelCase	,  __lowerCAmelCase				)				)
			# _get_minimal_slice_set is inclusive
			a__ :Optional[int]						  =    list(_flat_idx_to_idx(flat_end - 1	,  __lowerCAmelCase				)				)

			# Get an ordered list of slices to perform
			a__ :Union[str, Any]						  =    _get_minimal_slice_set(
			    __lowerCAmelCase	,  __lowerCAmelCase	,  __lowerCAmelCase	,  )

			a__ :Optional[Any]						  =    [t[s] for s in slices]

			return torch.cat([s.view((-1,) + t.shape[no_batch_dims:]				) for s in sliced_tensors]				)


def 				lowerCamelCase__	(      a	:   List[Any]	,  a	:   List[Any]	,  a	:   Any	,  a	:   int	,  a	:   List[Any] = False	,  a	:   List[str] = None	,  a	:   str = False	,  )    ->	Optional[Any]:





			"""simple docstring"""
			if not (len(__lowerCAmelCase				) > 0):
						raise ValueError("Must provide at least one input"				)

			a__ :int						  =    [shape[:no_batch_dims] for shape in _fetch_dims(__lowerCAmelCase				)]
			a__ :Optional[int]						  =    tuple([max(__lowerCAmelCase				) for s in zip(*__lowerCAmelCase				)]				)

			def _prep_inputs(a	:   Dict				) -> torch.Tensor:
						if not low_mem:
									if not sum(t.shape[:no_batch_dims]				) == no_batch_dims:
												a__ :Optional[Any]						  =    t.expand(orig_batch_dims + t.shape[no_batch_dims:]				)
									a__ :str						  =    t.reshape(-1	,  *t.shape[no_batch_dims:]				)
						else:
									a__ :List[str]						  =    t.expand(orig_batch_dims + t.shape[no_batch_dims:]				)
						return t

			a__ :Dict[str, Any]						  =    tensor_tree_map(_prep_inputs	,  __lowerCAmelCase				)
			a__ :List[Any]						  =    None
			if _out is not None:
						a__ :Optional[int]						  =    tensor_tree_map(lambda a				: t.view([-1] + list(t.shape[no_batch_dims:]				)				)	,  _out				)

			a__ :List[Any]						  =    1
			for d in orig_batch_dims:
						flat_batch_dim *= d

			a__ :List[Any]						  =    flat_batch_dim // chunk_size + (flat_batch_dim % chunk_size != 0)

			def _select_chunk(a	:   Dict				) -> torch.Tensor:
						return t[i : i + chunk_size] if t.shape[0] != 1 else t

			a__ :Tuple						  =    0
			a__ :Any						  =    prepped_outputs
			for _ in range(__lowerCAmelCase				):
						# Chunk the input
						if not low_mem:
									a__ :Optional[int]						  =    _select_chunk
						else:
									a__ :Tuple						  =    partial(
									    _chunk_slice	,  flat_start=__lowerCAmelCase	,  flat_end=min(__lowerCAmelCase	,  i + chunk_size				)	,  no_batch_dims=len(__lowerCAmelCase				)	,  )

						a__ :Dict[str, Any]						  =    tensor_tree_map(__lowerCAmelCase	,  __lowerCAmelCase				)

						# Run the layer on the chunk
						a__ :Optional[int]						  =    layer(**__lowerCAmelCase				)

						# Allocate space for the output
						if out is None:
									a__ :Optional[Any]						  =    tensor_tree_map(lambda a				: t.new_zeros((flat_batch_dim,) + t.shape[1:]				)	,  __lowerCAmelCase				)

						# Put the chunk in its pre-allocated space
						if isinstance(__lowerCAmelCase	,  __lowerCAmelCase				):

									def assign(a	:   Dict	,  a	:   Optional[Any]				) -> None:
												for k, v in da.items():
															if isinstance(__lowerCAmelCase	,  __lowerCAmelCase				):
																		assign(__lowerCAmelCase	,  da[k]				)
															else:
																		if _add_into_out:
																					v[i : i + chunk_size] += da[k]
																		else:
																					a__ :str						  =    da[k]

									assign(__lowerCAmelCase	,  __lowerCAmelCase				)
						elif isinstance(__lowerCAmelCase	,  __lowerCAmelCase				):
									for xa, xa in zip(__lowerCAmelCase	,  __lowerCAmelCase				):
												if _add_into_out:
															xa[i : i + chunk_size] += xa
												else:
															a__ :Any						  =    xa
						elif isinstance(__lowerCAmelCase	,  torch.Tensor				):
									if _add_into_out:
												out[i : i + chunk_size] += output_chunk
									else:
												a__ :int						  =    output_chunk
						else:
									raise ValueError("Not supported"				)

						i += chunk_size

			a__ :Union[str, Any]						  =    tensor_tree_map(lambda a				: t.view(orig_batch_dims + t.shape[1:]				)	,  __lowerCAmelCase				)

			return out







class 					lowerCAmelCase_ :




						def __init__(	self						:   Tuple   ,     __A						:   int = 512   ,     )      ->str:






									"""simple docstring"""






									a__ :str						  =    max_chunk_size
									a__ :Optional[int]						  =    None
									a__ :Optional[tuple]						  =    None




						def  _snake_case					(	self						:   Union[str, Any]   ,     __A						:   Callable   ,     __A						:   tuple   ,     __A						:   int						)      ->int:






									"""simple docstring"""






									logging.info("Tuning chunk size..."						)

									if min_chunk_size >= self.max_chunk_size:
												return min_chunk_size

									a__ :List[int]						  =    [2**l for l in range(int(math.log(self.max_chunk_size   ,     2						)						) + 1						)]
									a__ :List[str]						  =    [c for c in candidates if c > min_chunk_size]
									a__ :int						  =    [min_chunk_size] + candidates
									candidates[-1] += 4

									def test_chunk_size(__A						:   int						) -> bool:
												try:
															with torch.no_grad():
																		fn(*lowerCamelCase__   ,     chunk_size=lowerCamelCase__						)
															return True
												except RuntimeError:
															return False

									a__ :List[Any]						  =    0
									a__ :Tuple						  =    len(lowerCamelCase__						) - 1
									while i > min_viable_chunk_size_index:
												a__ :List[Any]						  =    test_chunk_size(candidates[i]						)
												if not viable:
															a__ :Union[str, Any]						  =    (min_viable_chunk_size_index + i) // 2
												else:
															a__ :Optional[Any]						  =    i
															a__ :int						  =    (i + len(lowerCamelCase__						) - 1) // 2

									return candidates[min_viable_chunk_size_index]




						def  _snake_case					(	self						:   Union[str, Any]   ,     __A						:   Iterable   ,     __A						:   Iterable						)      ->bool:






									"""simple docstring"""






									a__ :List[str]						  =    True
									for aa, aa in zip(lowerCamelCase__   ,     lowerCamelCase__						):
												assert type(lowerCamelCase__						) == type(lowerCamelCase__						)
												if isinstance(lowerCamelCase__   ,     (list, tuple)						):
															consistent &= self._compare_arg_caches(lowerCamelCase__   ,     lowerCamelCase__						)
												elif isinstance(lowerCamelCase__   ,     lowerCamelCase__						):
															a__ :str						  =    [v for _, v in sorted(aa.items()   ,     key=lambda __A						: x[0]						)]
															a__ :List[Any]						  =    [v for _, v in sorted(aa.items()   ,     key=lambda __A						: x[0]						)]
															consistent &= self._compare_arg_caches(lowerCamelCase__   ,     lowerCamelCase__						)
												else:
															consistent &= aa == aa

									return consistent






						def  _snake_case					(	self						:   List[str]   ,     __A						:   Callable   ,     __A						:   tuple   ,     __A						:   int   ,     )      ->int:






									"""simple docstring"""






									a__ :Optional[Any]						  =    True
									a__ :tuple						  =    tree_map(lambda __A						: a.shape if isinstance(lowerCamelCase__   ,     torch.Tensor						) else a   ,     lowerCamelCase__   ,     lowerCamelCase__						)
									if self.cached_arg_data is not None:
												# If args have changed shape/value, we need to re-tune
												assert len(self.cached_arg_data						) == len(lowerCamelCase__						)
												a__ :Dict						  =    self._compare_arg_caches(self.cached_arg_data   ,     lowerCamelCase__						)
									else:
												# Otherwise, we can reuse the precomputed value
												a__ :int						  =    False

									if not consistent:
												a__ :Any						  =    self._determine_favorable_chunk_size(
												    lowerCamelCase__   ,     lowerCamelCase__   ,     lowerCamelCase__   ,     )
												a__ :str						  =    arg_data

									assert self.cached_chunk_size is not None

									return self.cached_chunk_size






import datasets

from .evaluate import evaluate


snake_case__			 =						'''\
@inproceedings{Rajpurkar2016SQuAD10,
  title={SQuAD: 100, 000+ Questions for Machine Comprehension of Text},
  author={Pranav Rajpurkar and Jian Zhang and Konstantin Lopyrev and Percy Liang},
  booktitle={EMNLP},
  year={2016}
}
'''

snake_case__			 =						'''
This metric wrap the official scoring script for version 1 of the Stanford Question Answering Dataset (SQuAD).

Stanford Question Answering Dataset (SQuAD) is a reading comprehension dataset, consisting of questions posed by
crowdworkers on a set of Wikipedia articles, where the answer to every question is a segment of text, or span,
from the corresponding reading passage, or the question might be unanswerable.
'''

snake_case__			 =						'''
Computes SQuAD scores (F1 and EM).
Args:
    predictions: List of question-answers dictionaries with the following key-values:
        - \'id\': id of the question-answer pair as given in the references (see below)
        - \'prediction_text\': the text of the answer
    references: List of question-answers dictionaries with the following key-values:
        - \'id\': id of the question-answer pair (see above),
        - \'answers\': a Dict in the SQuAD dataset format
            {
                \'text\': list of possible texts for the answer, as a list of strings
                \'answer_start\': list of start positions for the answer, as a list of ints
            }
            Note that answer_start values are not taken into account to compute the metric.
Returns:
    \'exact_match\': Exact match (the normalized answer exactly match the gold answer)
    \'f1\': The F-score of predicted tokens versus the gold answer
Examples:

    >>> predictions = [{\'prediction_text\': \'1976\', \'id\': \'56e10a3be3433e1400422b22\'}]
    >>> references = [{\'answers\': {\'answer_start\': [97], \'text\': [\'1976\']}, \'id\': \'56e10a3be3433e1400422b22\'}]
    >>> squad_metric = datasets.load_metric("squad")
    >>> results = squad_metric.compute(predictions=predictions, references=references)
    >>> print(results)
    {\'exact_match\': 100.0, \'f1\': 100.0}
'''


@datasets.utils.file_utils.add_start_docstrings(_DESCRIPTION			,_KWARGS_DESCRIPTION)
class 					lowerCAmelCase_ (       datasets.Metric):




						def  _snake_case					(	self						:   List[str]						)      ->Union[str, Any]:






									"""simple docstring"""






									return datasets.MetricInfo(
									    description=_DESCRIPTION   ,     citation=_CITATION   ,     inputs_description=_KWARGS_DESCRIPTION   ,     features=datasets.Features(
									        {
									            "predictions": {"id": datasets.Value("string"						), "prediction_text": datasets.Value("string"						)},
									            "references": {
									                "id": datasets.Value("string"						),
									                "answers": datasets.features.Sequence(
									                    {
									                        "text": datasets.Value("string"						),
									                        "answer_start": datasets.Value("int32"						),
									                    }						),
									            },
									        }						)   ,     codebase_urls=["https://rajpurkar.github.io/SQuAD-explorer/"]   ,     reference_urls=["https://rajpurkar.github.io/SQuAD-explorer/"]   ,     )






						def  _snake_case					(	self						:   Dict   ,     __A						:   List[Any]   ,     __A						:   Optional[int]						)      ->str:






									"""simple docstring"""






									a__ :Optional[int]						  =    {prediction["id"]: prediction["prediction_text"] for prediction in predictions}
									a__ :Optional[Any]						  =    [
									    {
									        "paragraphs": [
									            {
									                "qas": [
									                    {
									                        "answers": [{"text": answer_text} for answer_text in ref["answers"]["text"]],
									                        "id": ref["id"],
									                    }
									                    for ref in references
									                ]
									            }
									        ]
									    }
									]
									a__ :Union[str, Any]						  =    evaluate(dataset=__A   ,     predictions=__A						)
									return score


