
'''simple docstring'''
import os

import pytest
import yaml

from datasets.features.features import Features, Value
from datasets.info import DatasetInfo, DatasetInfosDict







@pytest.mark.parametrize(
    """files"""			,     [
        ["""full:README.md""", """dataset_infos.json"""],
        ["""empty:README.md""", """dataset_infos.json"""],
        ["""dataset_infos.json"""],
        ["""full:README.md"""],
    ]			,     )
def 							A_     (						__SCREAMING_SNAKE_CASE						:  int			,     __SCREAMING_SNAKE_CASE						:  int )				->       Any:





  """simple docstring"""

  __A		:     int												=		tmp_path_factory.mktemp("""dset_infos_dir""" )
  if "full:README.md" in files:
    with open(dataset_infos_dir / """README.md"""			,     """w""" ) as f:
      f.write("""---\ndataset_info:\n  dataset_size: 42\n---""" )
  if "empty:README.md" in files:
    with open(dataset_infos_dir / """README.md"""			,     """w""" ) as f:
      f.write("""""" )
    # we want to support dataset_infos.json for backward compatibility
  if "dataset_infos.json" in files:
    with open(dataset_infos_dir / """dataset_infos.json"""			,     """w""" ) as f:
      f.write("""{\"default\": {\"dataset_size\": 42}}""" )
  __A		:     Optional[int]												=		DatasetInfosDict.from_directory(A_ )
  assert dataset_infos
  assert dataset_infos["default"].dataset_size == 42







@pytest.mark.parametrize(
    """dataset_info"""			,     [
        DatasetInfo(),
        DatasetInfo(
            description="""foo"""			,     features=Features({"""a""": Value("""int32""" )} )			,     builder_name="""builder"""			,     config_name="""config"""			,     version="""1.0.0"""			,     splits=[{"""name""": """train"""}]			,     download_size=42			,     ),
    ]			,     )
def 							A_     (						__SCREAMING_SNAKE_CASE						:  str			,     __SCREAMING_SNAKE_CASE						:  Optional[int] )				->       List[Any]:





  """simple docstring"""

  __A		:     Dict												=		str(A_ )
  dataset_info.write_to_directory(A_ )
  __A		:     Any												=		DatasetInfo.from_directory(A_ )
  assert dataset_info == reloaded
  assert os.path.exists(os.path.join(A_			,     """dataset_info.json""" ) )







def 							A_     (						)				->       Union[str, Any]:





  """simple docstring"""

  __A		:     Optional[Any]												=		DatasetInfo(
      description="""foo"""			,     citation="""bar"""			,     homepage="""https://foo.bar"""			,     license="""CC0"""			,     features=Features({"""a""": Value("""int32""" )} )			,     post_processed={}			,     supervised_keys=()			,     task_templates=[]			,     builder_name="""builder"""			,     config_name="""config"""			,     version="""1.0.0"""			,     splits=[{"""name""": """train""", """num_examples""": 42}]			,     download_checksums={}			,     download_size=1337			,     post_processing_size=442			,     dataset_size=1234			,     size_in_bytes=1337 + 442 + 1234			,     )
  __A		:     int												=		dataset_info._to_yaml_dict()
  assert sorted(A_ ) == sorted(DatasetInfo._INCLUDED_INFO_IN_YAML )
  for key in DatasetInfo._INCLUDED_INFO_IN_YAML:
    assert key in dataset_info_yaml_dict
    assert isinstance(dataset_info_yaml_dict[key]			,     (list, dict, int, str) )
  __A		:     Dict												=		yaml.safe_dump(A_ )
  __A		:     Union[str, Any]												=		yaml.safe_load(A_ )
  assert dataset_info_yaml_dict == reloaded







def 							A_     (						)				->       Dict:





  """simple docstring"""

  __A		:     int												=		DatasetInfo()
  __A		:     int												=		dataset_info._to_yaml_dict()
  assert dataset_info_yaml_dict == {}







@pytest.mark.parametrize(
    """dataset_infos_dict"""			,     [
        DatasetInfosDict(),
        DatasetInfosDict({"""default""": DatasetInfo()} ),
        DatasetInfosDict({"""my_config_name""": DatasetInfo()} ),
        DatasetInfosDict(
            {
                """default""": DatasetInfo(
                    description="""foo"""			,     features=Features({"""a""": Value("""int32""" )} )			,     builder_name="""builder"""			,     config_name="""config"""			,     version="""1.0.0"""			,     splits=[{"""name""": """train"""}]			,     download_size=42			,     )
            } ),
        DatasetInfosDict(
            {
                """v1""": DatasetInfo(dataset_size=42 ),
                """v2""": DatasetInfo(dataset_size=1337 ),
            } ),
    ]			,     )
def 							A_     (						__SCREAMING_SNAKE_CASE						:  Union[str, Any]			,     __SCREAMING_SNAKE_CASE						:  List[str] )				->       List[Any]:





  """simple docstring"""

  __A		:     Optional[Any]												=		str(A_ )
  dataset_infos_dict.write_to_directory(A_ )
  __A		:     Tuple												=		DatasetInfosDict.from_directory(A_ )

  # the config_name of the dataset_infos_dict take over the attribute
  for config_name, dataset_info in dataset_infos_dict.items():
    __A		:     str												=		config_name
    # the yaml representation doesn't include fields like description or citation
    # so we just test that we can recover what we can from the yaml
    __A		:     Any												=		DatasetInfo._from_yaml_dict(dataset_info._to_yaml_dict() )
  assert dataset_infos_dict == reloaded

  if dataset_infos_dict:
    assert os.path.exists(os.path.join(A_			,     """README.md""" ) )





'''simple docstring'''
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        A_     (						__SCREAMING_SNAKE_CASE						:  Union[dict, list, tuple, torch.Tensor] )				->       List[Tuple[int, ...]]:





  """simple docstring"""

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

  return shapes







@torch.jit.ignore
def        A_     (						__SCREAMING_SNAKE_CASE						:  int			,     __SCREAMING_SNAKE_CASE						:  Tuple[int, ...] )				->       Tuple[int, ...]:





  """simple docstring"""

  __A		:     List[str]												=		[]
  for d in reversed(__SCREAMING_SNAKE_CASE ):
    idx.append(flat_idx % d )
    __A		:     Any												=		flat_idx // d

  return tuple(reversed(__SCREAMING_SNAKE_CASE ) )







@torch.jit.ignore
def        A_     (						__SCREAMING_SNAKE_CASE						:  Sequence[int]			,     __SCREAMING_SNAKE_CASE						:  Sequence[int]			,     __SCREAMING_SNAKE_CASE						:  Sequence[int]			,     __SCREAMING_SNAKE_CASE						:  Optional[Sequence[bool]] = None			,     __SCREAMING_SNAKE_CASE						:  Optional[Sequence[bool]] = None			,     )				->       List[Tuple[slice, ...]]:





  """simple docstring"""

  def reduce_edge_list(__SCREAMING_SNAKE_CASE						:  List[bool] ) -> None:
    __A		:     Dict												=		True
    for i in range(len(__SCREAMING_SNAKE_CASE ) ):
      __A		:     List[str]												=		-1 * (i + 1)
      l[reversed_idx] &= tally
      __A		:     Optional[Any]												=		l[reversed_idx]

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

  # Base cases. Either start/end are empty and we're done, or the final,
  # one-dimensional tensor can be simply sliced
  if len(__SCREAMING_SNAKE_CASE ) == 0:
    return [()]
  elif len(__SCREAMING_SNAKE_CASE ) == 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(__SCREAMING_SNAKE_CASE			,     __SCREAMING_SNAKE_CASE ):
    if s == e:
      path_list.append(slice(__SCREAMING_SNAKE_CASE			,     s + 1 ) )
    else:
      break

  __A		:     Tuple[slice, ...]												=		tuple(__SCREAMING_SNAKE_CASE )
  __A		:     Dict												=		len(__SCREAMING_SNAKE_CASE )

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

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

    __A		:     List[Any]												=		start[divergence_idx]
    return tuple(
        path + (slice(__SCREAMING_SNAKE_CASE			,     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		:     Optional[int]												=		end[divergence_idx]
    return tuple(
        path + (slice(__SCREAMING_SNAKE_CASE			,     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		:     List[Any]												=		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        A_     (						__SCREAMING_SNAKE_CASE						:  torch.Tensor			,     __SCREAMING_SNAKE_CASE						:  int			,     __SCREAMING_SNAKE_CASE						:  int			,     __SCREAMING_SNAKE_CASE						:  int )				->       torch.Tensor:





  """simple docstring"""

  __A		:     List[Any]												=		t.shape[:no_batch_dims]
  __A		:     Dict												=		list(_flat_idx_to_idx(__SCREAMING_SNAKE_CASE			,     __SCREAMING_SNAKE_CASE ) )
  # _get_minimal_slice_set is inclusive
  __A		:     List[str]												=		list(_flat_idx_to_idx(flat_end - 1			,     __SCREAMING_SNAKE_CASE ) )

  # Get an ordered list of slices to perform
  __A		:     Optional[Any]												=		_get_minimal_slice_set(
      __SCREAMING_SNAKE_CASE			,     __SCREAMING_SNAKE_CASE			,     __SCREAMING_SNAKE_CASE			,     )

  __A		:     int												=		[t[s] for s in slices]

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







def        A_     (						__SCREAMING_SNAKE_CASE						:  Callable			,     __SCREAMING_SNAKE_CASE						:  Dict[str, Any]			,     __SCREAMING_SNAKE_CASE						:  int			,     __SCREAMING_SNAKE_CASE						:  int			,     __SCREAMING_SNAKE_CASE						:  bool = False			,     __SCREAMING_SNAKE_CASE						:  Any = None			,     __SCREAMING_SNAKE_CASE						:  bool = False			,     )				->       Any:





  """simple docstring"""

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

  __A		:     str												=		[shape[:no_batch_dims] for shape in _fetch_dims(__SCREAMING_SNAKE_CASE )]
  __A		:     Any												=		tuple([max(__SCREAMING_SNAKE_CASE ) for s in zip(*__SCREAMING_SNAKE_CASE )] )

  def _prep_inputs(__SCREAMING_SNAKE_CASE						:  torch.Tensor ) -> 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		:     Dict												=		t.reshape(-1			,     *t.shape[no_batch_dims:] )
    else:
      __A		:     str												=		t.expand(orig_batch_dims + t.shape[no_batch_dims:] )
    return t

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

  __A		:     Any												=		1
  for d in orig_batch_dims:
    flat_batch_dim *= d

  __A		:     Union[str, Any]												=		flat_batch_dim // chunk_size + (flat_batch_dim % chunk_size != 0)

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

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

    __A		:     Dict[str, Any]												=		tensor_tree_map(__SCREAMING_SNAKE_CASE			,     __SCREAMING_SNAKE_CASE )

    # Run the layer on the chunk
    __A		:     List[Any]												=		layer(**__SCREAMING_SNAKE_CASE )

    # Allocate space for the output
    if out is None:
      __A		:     str												=		tensor_tree_map(lambda __SCREAMING_SNAKE_CASE : t.new_zeros((flat_batch_dim,) + t.shape[1:] )			,     __SCREAMING_SNAKE_CASE )

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

      def assign(__SCREAMING_SNAKE_CASE						:  dict			,     __SCREAMING_SNAKE_CASE						:  dict ) -> None:
        for k, v in da.items():
          if isinstance(__SCREAMING_SNAKE_CASE			,     __SCREAMING_SNAKE_CASE ):
            assign(__SCREAMING_SNAKE_CASE			,     da[k] )
          else:
            if _add_into_out:
              v[i : i + chunk_size] += da[k]
            else:
              __A		:     int												=		da[k]

      assign(__SCREAMING_SNAKE_CASE			,     __SCREAMING_SNAKE_CASE )
    elif isinstance(__SCREAMING_SNAKE_CASE			,     __SCREAMING_SNAKE_CASE ):
      for xa, xa in zip(__SCREAMING_SNAKE_CASE			,     __SCREAMING_SNAKE_CASE ):
        if _add_into_out:
          xa[i : i + chunk_size] += xa
        else:
          __A		:     int												=		xa
    elif isinstance(__SCREAMING_SNAKE_CASE			,     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		:     Optional[Any]												=		tensor_tree_map(lambda __SCREAMING_SNAKE_CASE : t.view(orig_batch_dims + t.shape[1:] )			,     __SCREAMING_SNAKE_CASE )

  return out







class  __A				:
 def __init__(   self			:    Any				,   lowerCamelCase			:    int = 5_12				,   ):



   """simple docstring"""





   __A		:     Optional[Any]												=		max_chunk_size
   __A		:     Optional[int]												=		None
   __A		:     Optional[tuple]												=		None
 def 		lowercase_(   self			:    Union[str, Any]				,   lowerCamelCase			:    Callable				,   lowerCamelCase			:    tuple				,   lowerCamelCase			:    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		:     List[str]												=		[min_chunk_size] + candidates
   candidates[-1] += 4

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

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

   return candidates[min_viable_chunk_size_index]
 def 		lowercase_(   self			:    Dict				,   lowerCamelCase			:    Iterable				,   lowerCamelCase			:    Iterable       ):



   """simple docstring"""





   __A		:     Optional[int]												=		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		:     Dict												=		[v for _, v in sorted(aa.items()				,   key=lambda lowerCamelCase       : x[0]       )]
       __A		:     Any												=		[v for _, v in sorted(aa.items()				,   key=lambda lowerCamelCase       : x[0]       )]
       consistent &= self._compare_arg_caches(lowerCamelCase				,   lowerCamelCase       )
     else:
       consistent &= aa == aa

   return consistent






 def 		lowercase_(   self			:    Optional[int]				,   lowerCamelCase			:    Callable				,   lowerCamelCase			:    tuple				,   lowerCamelCase			:    int				,   ):



   """simple docstring"""





   __A		:     str												=		True
   __A		:     tuple												=		tree_map(lambda lowerCamelCase       : 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		:     Optional[Any]												=		self._compare_arg_caches(self.cached_arg_data				,   lowerCamelCase       )
   else:
     # Otherwise, we can reuse the precomputed value
     __A		:     List[str]												=		False

   if not consistent:
     __A		:     List[Any]												=		self._determine_favorable_chunk_size(
         lowerCamelCase				,   lowerCamelCase				,   lowerCamelCase				,   )
     __A		:     Optional[int]												=		arg_data

   assert self.cached_chunk_size is not None

   return self.cached_chunk_size



