
'''simple docstring'''
import json
import os
from functools import lru_cache
from typing import Dict, List, Optional, Tuple, Union

import regex as re

from ...tokenization_utils import AddedToken, PreTrainedTokenizer
from ...tokenization_utils_base import BatchEncoding, EncodedInput
from ...utils import PaddingStrategy, logging


A__   :		Union[str, Any] 		=logging.get_logger(__name__)


A__   :		List[str] 		={'vocab_file': 'vocab.json', 'merges_file': 'merges.txt'}

# See all LED models at https://huggingface.co/models?filter=LED
A__   :		str 		={
    'vocab_file': {
        'allenai/led-base-16384': 'https://huggingface.co/allenai/led-base-16384/resolve/main/vocab.json',
    },
    'merges_file': {
        'allenai/led-base-16384': 'https://huggingface.co/allenai/led-base-16384/resolve/main/merges.txt',
    },
    'tokenizer_file': {
        'allenai/led-base-16384': 'https://huggingface.co/allenai/led-base-16384/resolve/main/tokenizer.json',
    },
}

A__   :		List[Any] 		={
    'allenai/led-base-16384': 16_384,
}







@lru_cache()
# Copied from transformers.models.bart.tokenization_bart.bytes_to_unicode
def 							A_     (						)				->       Tuple:





  """simple docstring"""

  __A		:     Optional[int]												=		(
      list(range(ord("""!""" )			,     ord("""~""" ) + 1 ) ) + list(range(ord("""¡""" )			,     ord("""¬""" ) + 1 ) ) + list(range(ord("""®""" )			,     ord("""ÿ""" ) + 1 ) )
  )
  __A		:     List[Any]												=		bs[:]
  __A		:     Union[str, Any]												=		0
  for b in range(2**8 ):
    if b not in bs:
      bs.append(__SCREAMING_SNAKE_CASE )
      cs.append(2**8 + n )
      n += 1
  __A		:     int												=		[chr(__SCREAMING_SNAKE_CASE ) for n in cs]
  return dict(zip(__SCREAMING_SNAKE_CASE			,     __SCREAMING_SNAKE_CASE ) )







def 							A_     (						__SCREAMING_SNAKE_CASE						:  str )				->       Tuple:





  """simple docstring"""

  __A		:     str												=		set()
  __A		:     int												=		word[0]
  for char in word[1:]:
    pairs.add((prev_char, char) )
    __A		:     Optional[Any]												=		char
  return pairs







class  __A				( _SCREAMING_SNAKE_CASE			):
 lowerCamelCase				       =VOCAB_FILES_NAMES
 lowerCamelCase				       =PRETRAINED_VOCAB_FILES_MAP
 lowerCamelCase				       =PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
 lowerCamelCase				       =['''input_ids''', '''attention_mask''']
 def __init__(   self			:    Any				,   lowerCamelCase			:    Dict				,   lowerCamelCase			:    List[str]				,   lowerCamelCase			:    Dict="replace"				,   lowerCamelCase			:    int="<s>"				,   lowerCamelCase			:    str="</s>"				,   lowerCamelCase			:    List[str]="</s>"				,   lowerCamelCase			:    Optional[Any]="<s>"				,   lowerCamelCase			:    Any="<unk>"				,   lowerCamelCase			:    str="<pad>"				,   lowerCamelCase			:    Any="<mask>"				,   lowerCamelCase			:    Optional[int]=False				,   **lowerCamelCase			:    str				,   ):



   """simple docstring"""





   __A		:     List[Any]												=		AddedToken(lowerCamelCase				,   lstrip=lowerCamelCase				,   rstrip=lowerCamelCase       ) if isinstance(lowerCamelCase				,   lowerCamelCase       ) else bos_token
   __A		:     Any												=		AddedToken(lowerCamelCase				,   lstrip=lowerCamelCase				,   rstrip=lowerCamelCase       ) if isinstance(lowerCamelCase				,   lowerCamelCase       ) else eos_token
   __A		:     int												=		AddedToken(lowerCamelCase				,   lstrip=lowerCamelCase				,   rstrip=lowerCamelCase       ) if isinstance(lowerCamelCase				,   lowerCamelCase       ) else sep_token
   __A		:     Any												=		AddedToken(lowerCamelCase				,   lstrip=lowerCamelCase				,   rstrip=lowerCamelCase       ) if isinstance(lowerCamelCase				,   lowerCamelCase       ) else cls_token
   __A		:     List[Any]												=		AddedToken(lowerCamelCase				,   lstrip=lowerCamelCase				,   rstrip=lowerCamelCase       ) if isinstance(lowerCamelCase				,   lowerCamelCase       ) else unk_token
   __A		:     List[Any]												=		AddedToken(lowerCamelCase				,   lstrip=lowerCamelCase				,   rstrip=lowerCamelCase       ) if isinstance(lowerCamelCase				,   lowerCamelCase       ) else pad_token

   # Mask token behave like a normal word, i.e. include the space before it
   __A		:     List[str]												=		AddedToken(lowerCamelCase				,   lstrip=lowerCamelCase				,   rstrip=lowerCamelCase       ) if isinstance(lowerCamelCase				,   lowerCamelCase       ) else mask_token

   super().__init__(
       errors=lowerCamelCase				,   bos_token=lowerCamelCase				,   eos_token=lowerCamelCase				,   unk_token=lowerCamelCase				,   sep_token=lowerCamelCase				,   cls_token=lowerCamelCase				,   pad_token=lowerCamelCase				,   mask_token=lowerCamelCase				,   add_prefix_space=lowerCamelCase				,   **lowerCamelCase				,   )

   with open(lowerCamelCase				,   encoding="""utf-8"""       ) as vocab_handle:
     __A		:     Optional[Any]												=		json.load(lowerCamelCase       )
   __A		:     Dict												=		{v: k for k, v in self.encoder.items()}
   __A		:     List[Any]												=		errors  # how to handle errors in decoding
   __A		:     Optional[int]												=		bytes_to_unicode()
   __A		:     str												=		{v: k for k, v in self.byte_encoder.items()}
   with open(lowerCamelCase				,   encoding="""utf-8"""       ) as merges_handle:
     __A		:     str												=		merges_handle.read().split("""\n"""       )[1:-1]
   __A		:     Tuple												=		[tuple(merge.split()       ) for merge in bpe_merges]
   __A		:     List[Any]												=		dict(zip(lowerCamelCase				,   range(len(lowerCamelCase       )       )       )       )
   __A		:     List[str]												=		{}
   __A		:     List[str]												=		add_prefix_space

   # Should have added re.IGNORECASE so BPE merges can happen for capitalized versions of contractions
   __A		:     Any												=		re.compile(r"""'s|'t|'re|'ve|'m|'ll|'d| ?\p{L}+| ?\p{N}+| ?[^\s\p{L}\p{N}]+|\s+(?!\S)|\s+"""       )
 @property
 # Copied from transformers.models.bart.tokenization_bart.BartTokenizer.vocab_size
 def 		lowercase_(   self			:    Any       ):



   """simple docstring"""





   return len(self.encoder       )
 def 		lowercase_(   self			:    Optional[int]       ):



   """simple docstring"""





   return dict(self.encoder				,   **self.added_tokens_encoder       )
 def 		lowercase_(   self			:    str				,   lowerCamelCase			:    Optional[int]       ):



   """simple docstring"""





   if token in self.cache:
     return self.cache[token]
   __A		:     Optional[Any]												=		tuple(lowerCamelCase       )
   __A		:     Dict												=		get_pairs(lowerCamelCase       )

   if not pairs:
     return token

   while True:
     __A		:     str												=		min(lowerCamelCase				,   key=lambda lowerCamelCase       : self.bpe_ranks.get(lowerCamelCase				,   float("""inf"""       )       )       )
     if bigram not in self.bpe_ranks:
       break
     __A		:     Optional[Any]												=		bigram
     __A		:     List[Any]												=		[]
     __A		:     Union[str, Any]												=		0
     while i < len(lowerCamelCase       ):
       try:
         __A		:     str												=		word.index(lowerCamelCase				,   lowerCamelCase       )
       except ValueError:
         new_word.extend(word[i:]       )
         break
       else:
         new_word.extend(word[i:j]       )
         __A		:     Union[str, Any]												=		j

       if word[i] == first and i < len(lowerCamelCase       ) - 1 and word[i + 1] == second:
         new_word.append(first + second       )
         i += 2
       else:
         new_word.append(word[i]       )
         i += 1
     __A		:     Dict												=		tuple(lowerCamelCase       )
     __A		:     List[Any]												=		new_word
     if len(lowerCamelCase       ) == 1:
       break
     else:
       __A		:     Optional[Any]												=		get_pairs(lowerCamelCase       )
   __A		:     Optional[int]												=		""" """.join(lowerCamelCase       )
   __A		:     str												=		word
   return word
 def 		lowercase_(   self			:    str				,   lowerCamelCase			:    Optional[int]       ):



   """simple docstring"""





   __A		:     List[str]												=		[]
   for token in re.findall(self.pat				,   lowerCamelCase       ):
     __A		:     Optional[Any]												=		"""""".join(
         self.byte_encoder[b] for b in token.encode("""utf-8"""       )       )  # Maps all our bytes to unicode strings, avoiding control tokens of the BPE (spaces in our case)
     bpe_tokens.extend(bpe_token for bpe_token in self.bpe(lowerCamelCase       ).split(""" """       )       )
   return bpe_tokens
 def 		lowercase_(   self			:    Dict				,   lowerCamelCase			:    Tuple       ):



   """simple docstring"""





   return self.encoder.get(lowerCamelCase				,   self.encoder.get(self.unk_token       )       )
 def 		lowercase_(   self			:    str				,   lowerCamelCase			:    Optional[Any]       ):



   """simple docstring"""





   return self.decoder.get(lowerCamelCase       )
 def 		lowercase_(   self			:    Tuple				,   lowerCamelCase			:    int       ):



   """simple docstring"""





   __A		:     Any												=		"""""".join(lowerCamelCase       )
   __A		:     int												=		bytearray([self.byte_decoder[c] for c in text]       ).decode("""utf-8"""				,   errors=self.errors       )
   return text
 def 		lowercase_(   self			:    List[str]				,   lowerCamelCase			:    str				,   lowerCamelCase			:    Optional[str] = None       ):



   """simple docstring"""





   if not os.path.isdir(lowerCamelCase       ):
     logger.error(f"Vocabulary path ({save_directory}) should be a directory"       )
     return
   __A		:     Any												=		os.path.join(
       lowerCamelCase				,   (filename_prefix + """-""" if filename_prefix else """""") + VOCAB_FILES_NAMES["""vocab_file"""]       )
   __A		:     Optional[Any]												=		os.path.join(
       lowerCamelCase				,   (filename_prefix + """-""" if filename_prefix else """""") + VOCAB_FILES_NAMES["""merges_file"""]       )

   with open(lowerCamelCase				,   """w"""				,   encoding="""utf-8"""       ) as f:
     f.write(json.dumps(self.encoder				,   indent=2				,   sort_keys=lowerCamelCase				,   ensure_ascii=lowerCamelCase       ) + """\n"""       )

   __A		:     str												=		0
   with open(lowerCamelCase				,   """w"""				,   encoding="""utf-8"""       ) as writer:
     writer.write("""#version: 0.2\n"""       )
     for bpe_tokens, token_index in sorted(self.bpe_ranks.items()				,   key=lambda lowerCamelCase       : kv[1]       ):
       if index != token_index:
         logger.warning(
             f"Saving vocabulary to {merge_file}: BPE merge indices are not consecutive."
             """ Please check that the tokenizer is not corrupted!"""       )
         __A		:     Any												=		token_index
       writer.write(""" """.join(lowerCamelCase       ) + """\n"""       )
       index += 1

   return vocab_file, merge_file
 def 		lowercase_(   self			:    str				,   lowerCamelCase			:    List[int]				,   lowerCamelCase			:    Optional[List[int]] = None       ):



   """simple docstring"""





   if token_ids_a is None:
     return [self.cls_token_id] + token_ids_a + [self.sep_token_id]
   __A		:     Tuple												=		[self.cls_token_id]
   __A		:     str												=		[self.sep_token_id]
   return cls + token_ids_a + sep + sep + token_ids_a + sep
 def 		lowercase_(   self			:    List[str]				,   lowerCamelCase			:    List[int]				,   lowerCamelCase			:    Optional[List[int]] = None				,   lowerCamelCase			:    bool = False       ):



   """simple docstring"""





   if already_has_special_tokens:
     return super().get_special_tokens_mask(
         token_ids_a=lowerCamelCase				,   token_ids_a=lowerCamelCase				,   already_has_special_tokens=lowerCamelCase       )

   if token_ids_a is None:
     return [1] + ([0] * len(lowerCamelCase       )) + [1]
   return [1] + ([0] * len(lowerCamelCase       )) + [1, 1] + ([0] * len(lowerCamelCase       )) + [1]
 def 		lowercase_(   self			:    List[str]				,   lowerCamelCase			:    List[int]				,   lowerCamelCase			:    Optional[List[int]] = None       ):



   """simple docstring"""





   __A		:     int												=		[self.sep_token_id]
   __A		:     List[str]												=		[self.cls_token_id]

   if token_ids_a is None:
     return len(cls + token_ids_a + sep       ) * [0]
   return len(cls + token_ids_a + sep + sep + token_ids_a + sep       ) * [0]
 def 		lowercase_(   self			:    Dict				,   lowerCamelCase			:    int				,   lowerCamelCase			:    Dict=False				,   **lowerCamelCase			:    List[Any]       ):



   """simple docstring"""





   __A		:     Union[str, Any]												=		kwargs.pop("""add_prefix_space"""				,   self.add_prefix_space       )
   if (is_split_into_words or add_prefix_space) and (len(lowerCamelCase       ) > 0 and not text[0].isspace()):
     __A		:     Dict												=		""" """ + text
   return (text, kwargs)






 def 		lowercase_(   self			:    Optional[int]				,   lowerCamelCase			:    Union[Dict[str, EncodedInput], BatchEncoding]				,   lowerCamelCase			:    Optional[int] = None				,   lowerCamelCase			:    PaddingStrategy = PaddingStrategy.DO_NOT_PAD				,   lowerCamelCase			:    Optional[int] = None				,   lowerCamelCase			:    Optional[bool] = None				,   ):



   """simple docstring"""





   __A		:     str												=		super()._pad(
       encoded_inputs=lowerCamelCase				,   max_length=lowerCamelCase				,   padding_strategy=lowerCamelCase				,   pad_to_multiple_of=lowerCamelCase				,   return_attention_mask=lowerCamelCase				,   )

   # Load from model defaults
   if return_attention_mask is None:
     __A		:     List[Any]												=		"""attention_mask""" in self.model_input_names

   if return_attention_mask and "global_attention_mask" in encoded_inputs:
     __A		:     Tuple												=		encoded_inputs[self.model_input_names[0]]
     # `global_attention_mask` need to have the same length as other (sequential) inputs.
     __A		:     List[Any]												=		len(encoded_inputs["""global_attention_mask"""]       ) != len(lowerCamelCase       )

     if needs_to_be_padded:
       __A		:     str												=		len(lowerCamelCase       ) - len(encoded_inputs["""global_attention_mask"""]       )

       if self.padding_side == "right":
         # Use `-1` since `0` in `global_attention_mask` means `local attention` instead of `not to attend`
         __A		:     List[Any]												=		(
             encoded_inputs["""global_attention_mask"""] + [-1] * difference
         )
       elif self.padding_side == "left":
         __A		:     Any												=		[-1] * difference + encoded_inputs[
             """global_attention_mask"""
         ]
       else:
         raise ValueError("""Invalid padding strategy:""" + str(self.padding_side       )       )

   return encoded_inputs





'''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



