


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__    (		snake_case__					:							Union[dict, list, tuple, torch.Tensor]     ):
  _UpperCAmelCase		:							str							=     []
  if isinstance(snake_case__						,   snake_case__     ):
    for v in tree.values():
      shapes.extend(_fetch_dims(snake_case__     )     )
  elif isinstance(snake_case__						,   (list, tuple)     ):
    for t in tree:
      shapes.extend(_fetch_dims(snake_case__     )     )
  elif isinstance(snake_case__						,   torch.Tensor     ):
    shapes.append(tree.shape     )
  else:
    raise ValueError("""Not supported"""     )

  return shapes




@torch.jit.ignore
def    a__    (		snake_case__					:							int						,   snake_case__					:							Tuple[int, ...]     ):
  _UpperCAmelCase		:							str							=     []
  for d in reversed(snake_case__     ):
    idx.append(flat_idx % d     )
    _UpperCAmelCase		:							Any							=     flat_idx // d

  return tuple(reversed(snake_case__     )     )




@torch.jit.ignore
def    a__    (		snake_case__					:							Sequence[int]						,   snake_case__					:							Sequence[int]						,   snake_case__					:							Sequence[int]						,   snake_case__					:							Optional[Sequence[bool]] = None						,   snake_case__					:							Optional[Sequence[bool]] = None						,   ):

  # 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(snake_case__					:							List[bool]     ) -> None:
    _UpperCAmelCase		:							List[Any]							=     True
    for i in range(len(snake_case__     )     ):
      _UpperCAmelCase		:							str							=     -1 * (i + 1)
      l[reversed_idx] &= tally
      _UpperCAmelCase		:							Any							=     l[reversed_idx]

  if start_edges is None:
    _UpperCAmelCase		:							List[str]							=     [s == 0 for s in start]
    reduce_edge_list(snake_case__     )
  if end_edges is None:
    _UpperCAmelCase		:							int							=     [e == (d - 1) for e, d in zip(snake_case__						,   snake_case__     )]
    reduce_edge_list(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(snake_case__     ) == 0:
    return [()]
  elif len(snake_case__     ) == 1:
    return [(slice(start[0]						,   end[0] + 1     ),)]

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

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

  _UpperCAmelCase		:							Tuple[slice, ...]							=     tuple(snake_case__     )
  _UpperCAmelCase		:							Any							=     len(snake_case__     )

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

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

    _UpperCAmelCase		:							Union[str, Any]							=     start[divergence_idx]
    return tuple(
        path + (slice(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

    _UpperCAmelCase		:							int							=     end[divergence_idx]
    return tuple(
        path + (slice(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()     )
    _UpperCAmelCase		:							List[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    a__    (		snake_case__					:							torch.Tensor						,   snake_case__					:							int						,   snake_case__					:							int						,   snake_case__					:							int     ):
  _UpperCAmelCase		:							List[Any]							=     t.shape[:no_batch_dims]
  _UpperCAmelCase		:							Optional[Any]							=     list(_flat_idx_to_idx(snake_case__						,   snake_case__     )     )
  # _get_minimal_slice_set is inclusive
  _UpperCAmelCase		:							Dict							=     list(_flat_idx_to_idx(flat_end - 1						,   snake_case__     )     )

  # Get an ordered list of slices to perform
  _UpperCAmelCase		:							List[str]							=     _get_minimal_slice_set(
      snake_case__						,   snake_case__						,   snake_case__						,   )

  _UpperCAmelCase		:							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    a__    (		snake_case__					:							Callable						,   snake_case__					:							Dict[str, Any]						,   snake_case__					:							int						,   snake_case__					:							int						,   snake_case__					:							bool = False						,   snake_case__					:							Any = None						,   snake_case__					:							bool = False						,   ):
  if not (len(snake_case__     ) > 0):
    raise ValueError("""Must provide at least one input"""     )

  _UpperCAmelCase		:							Optional[int]							=     [shape[:no_batch_dims] for shape in _fetch_dims(snake_case__     )]
  _UpperCAmelCase		:							List[Any]							=     tuple([max(snake_case__     ) for s in zip(*snake_case__     )]     )

  def _prep_inputs(snake_case__					:							torch.Tensor     ) -> torch.Tensor:
    if not low_mem:
      if not sum(t.shape[:no_batch_dims]     ) == no_batch_dims:
        _UpperCAmelCase		:							Optional[int]							=     t.expand(orig_batch_dims + t.shape[no_batch_dims:]     )
      _UpperCAmelCase		:							Tuple							=     t.reshape(-1						,   *t.shape[no_batch_dims:]     )
    else:
      _UpperCAmelCase		:							Dict							=     t.expand(orig_batch_dims + t.shape[no_batch_dims:]     )
    return t

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

  _UpperCAmelCase		:							int							=     1
  for d in orig_batch_dims:
    flat_batch_dim *= d

  _UpperCAmelCase		:							Optional[int]							=     flat_batch_dim // chunk_size + (flat_batch_dim % chunk_size != 0)

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

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

    _UpperCAmelCase		:							Dict[str, Any]							=     tensor_tree_map(snake_case__						,   snake_case__     )

    # Run the layer on the chunk
    _UpperCAmelCase		:							List[Any]							=     layer(**snake_case__     )

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

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

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

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

    i += chunk_size

  _UpperCAmelCase		:							Optional[Any]							=     tensor_tree_map(lambda snake_case__     : t.view(orig_batch_dims + t.shape[1:]     )						,   snake_case__     )

  return out





class 		_SCREAMING_SNAKE_CASE     :


     def __init__(  self							,					A_ = 5_12							,					):
       _UpperCAmelCase		:							Tuple							=     max_chunk_size
       _UpperCAmelCase		:							Optional[int]							=     None
       _UpperCAmelCase		:							Optional[tuple]							=     None


     def        __snake_case(  self							,					A_							,					A_							,					A_    ):
       logging.info("""Tuning chunk size..."""    )

       if min_chunk_size >= self.max_chunk_size:
         return min_chunk_size

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

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

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

       return candidates[min_viable_chunk_size_index]


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

       return consistent




     def        __snake_case(  self							,					A_							,					A_							,					A_							,					):
       _UpperCAmelCase		:							Dict							=     True
       _UpperCAmelCase		:							tuple							=     tree_map(lambda A_    : a.shape if isinstance(A_							,					torch.Tensor    ) else a							,					A_							,					A_    )
       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(A_    )
         _UpperCAmelCase		:							Optional[Any]							=     self._compare_arg_caches(self.cached_arg_data							,					A_    )
       else:
         # Otherwise, we can reuse the precomputed value
         _UpperCAmelCase		:							Any							=     False

       if not consistent:
         _UpperCAmelCase		:							Dict							=     self._determine_favorable_chunk_size(
             A_							,					A_							,					A_							,					)
         _UpperCAmelCase		:							Any							=     arg_data

       assert self.cached_chunk_size is not None

       return self.cached_chunk_size









from __future__ import annotations

from typing import Dict

from ...configuration_utils import PretrainedConfig


SCREAMING_SNAKE_CASE__					:     List[Any]       =     {
    'susnato/ernie-m-base_pytorch': 'https://huggingface.co/susnato/ernie-m-base_pytorch/blob/main/config.json',
    'susnato/ernie-m-large_pytorch': 'https://huggingface.co/susnato/ernie-m-large_pytorch/blob/main/config.json',
}


class 		_SCREAMING_SNAKE_CASE     (						A     ):
     __SCREAMING_SNAKE_CASE     					=    '''ernie_m'''
     __SCREAMING_SNAKE_CASE     					=    {"dropout": "classifier_dropout", "num_classes": "num_labels"}


     def __init__(  self							,					A_ = 25_00_02							,					A_ = 7_68							,					A_ = 12							,					A_ = 12							,					A_ = 30_72							,					A_ = "gelu"							,					A_ = 0.1							,					A_ = 0.1							,					A_ = 5_14							,					A_ = 0.0_2							,					A_ = 1							,					A_ = 1e-05							,					A_=None							,					A_=False							,					A_=0.0							,					**A_							,					):
       super().__init__(pad_token_id=A_							,					**A_    )
       _UpperCAmelCase		:							Tuple							=     vocab_size
       _UpperCAmelCase		:							List[str]							=     hidden_size
       _UpperCAmelCase		:							Optional[int]							=     num_hidden_layers
       _UpperCAmelCase		:							Tuple							=     num_attention_heads
       _UpperCAmelCase		:							Dict							=     intermediate_size
       _UpperCAmelCase		:							List[Any]							=     hidden_act
       _UpperCAmelCase		:							Any							=     hidden_dropout_prob
       _UpperCAmelCase		:							Any							=     attention_probs_dropout_prob
       _UpperCAmelCase		:							Tuple							=     max_position_embeddings
       _UpperCAmelCase		:							Dict							=     initializer_range
       _UpperCAmelCase		:							Optional[Any]							=     layer_norm_eps
       _UpperCAmelCase		:							str							=     classifier_dropout
       _UpperCAmelCase		:							Any							=     is_decoder
       _UpperCAmelCase		:							Union[str, Any]							=     act_dropout





