
'''simple docstring'''
import dataclasses
import re
import string
from typing import Any, Dict, Iterator, List, Mapping, Optional, Sequence, Tuple

import numpy as np

from . import residue_constants


A__   :		Optional[Any] 		=Mapping[str, np.ndarray]
A__   :		Union[str, Any] 		=Mapping[str, Any]  # Is a nested dict.
A__   :		int 		=0.0_1







@dataclasses.dataclass(frozen=_SCREAMING_SNAKE_CASE			)
class  __A				:
 lowerCamelCase				       =42  # [num_res, num_atom_type, 3]

 # Amino-acid type for each residue represented as an integer between 0 and
 # 20, where 20 is 'X'.
 lowerCamelCase				       =42  # [num_res]

 # Binary float mask to indicate presence of a particular atom. 1.0 if an atom
 # is present and 0.0 if not. This should be used for loss masking.
 lowerCamelCase				       =42  # [num_res, num_atom_type]

 # Residue index as used in PDB. It is not necessarily continuous or 0-indexed.
 lowerCamelCase				       =42  # [num_res]

 # B-factors, or temperature factors, of each residue (in sq. angstroms units),
 # representing the displacement of the residue from its ground truth mean
 # value.
 lowerCamelCase				       =42  # [num_res, num_atom_type]

 # Chain indices for multi-chain predictions
 lowerCamelCase				       =None

 # Optional remark about the protein. Included as a comment in output PDB
 # files
 lowerCamelCase				       =None

 # Templates used to generate this protein (prediction-only)
 lowerCamelCase				       =None

 # Chain corresponding to each parent
 lowerCamelCase				       =None







def 							A_     (						__SCREAMING_SNAKE_CASE						:  str )				->       Protein:





  """simple docstring"""

  __A		:     Optional[int]												=		R"""(\[[A-Z]+\]\n)"""
  __A		:     List[str]												=		[tag.strip() for tag in re.split(__SCREAMING_SNAKE_CASE			,     __SCREAMING_SNAKE_CASE ) if len(__SCREAMING_SNAKE_CASE ) > 0]
  __A		:     Iterator[Tuple[str, List[str]]]												=		zip(tags[0::2]			,     [l.split("""\n""" ) for l in tags[1::2]] )

  __A		:     List[str]												=		["N", "CA", "C"]
  __A		:     Dict												=		None
  __A		:     List[str]												=		None
  __A		:     int												=		None
  for g in groups:
    if "[PRIMARY]" == g[0]:
      __A		:     List[str]												=		g[1][0].strip()
      for i in range(len(__SCREAMING_SNAKE_CASE ) ):
        if seq[i] not in residue_constants.restypes:
          __A		:     Tuple												=		"""X"""  # FIXME: strings are immutable
      __A		:     str												=		np.array(
          [residue_constants.restype_order.get(__SCREAMING_SNAKE_CASE			,     residue_constants.restype_num ) for res_symbol in seq] )
    elif "[TERTIARY]" == g[0]:
      __A		:     List[List[float]]												=		[]
      for axis in range(3 ):
        tertiary.append(list(map(__SCREAMING_SNAKE_CASE			,     g[1][axis].split() ) ) )
      __A		:     Any												=		np.array(__SCREAMING_SNAKE_CASE )
      __A		:     str												=		np.zeros((len(tertiary[0] ) // 3, residue_constants.atom_type_num, 3) ).astype(np.floataa )
      for i, atom in enumerate(__SCREAMING_SNAKE_CASE ):
        __A		:     str												=		np.transpose(tertiary_np[:, i::3] )
      atom_positions *= PICO_TO_ANGSTROM
    elif "[MASK]" == g[0]:
      __A		:     Optional[Any]												=		np.array(list(map({"""-""": 0, """+""": 1}.get			,     g[1][0].strip() ) ) )
      __A		:     str												=		np.zeros(
          (
              len(__SCREAMING_SNAKE_CASE ),
              residue_constants.atom_type_num,
          ) ).astype(np.floataa )
      for i, atom in enumerate(__SCREAMING_SNAKE_CASE ):
        __A		:     Optional[Any]												=		1
      atom_mask *= mask[..., None]

  assert aatype is not None

  return Protein(
      atom_positions=__SCREAMING_SNAKE_CASE			,     atom_mask=__SCREAMING_SNAKE_CASE			,     aatype=__SCREAMING_SNAKE_CASE			,     residue_index=np.arange(len(__SCREAMING_SNAKE_CASE ) )			,     b_factors=__SCREAMING_SNAKE_CASE			,     )







def 							A_     (						__SCREAMING_SNAKE_CASE						:  Protein			,     __SCREAMING_SNAKE_CASE						:  int = 0 )				->       List[str]:





  """simple docstring"""

  __A		:     List[str]												=		[]

  __A		:     Optional[Any]												=		prot.remark
  if remark is not None:
    pdb_headers.append(F"REMARK {remark}" )

  __A		:     int												=		prot.parents
  __A		:     Optional[Any]												=		prot.parents_chain_index
  if parents is not None and parents_chain_index is not None:
    __A		:     Optional[Any]												=		[p for i, p in zip(__SCREAMING_SNAKE_CASE			,     __SCREAMING_SNAKE_CASE ) if i == chain_id]

  if parents is None or len(__SCREAMING_SNAKE_CASE ) == 0:
    __A		:     Union[str, Any]												=		["""N/A"""]

  pdb_headers.append(F"PARENT {' '.join(__SCREAMING_SNAKE_CASE )}" )

  return pdb_headers







def 							A_     (						__SCREAMING_SNAKE_CASE						:  Protein			,     __SCREAMING_SNAKE_CASE						:  str )				->       str:





  """simple docstring"""

  __A		:     List[str]												=		[]
  __A		:     Optional[int]												=		pdb_str.split("""\n""" )

  __A		:     Optional[int]												=		prot.remark
  if remark is not None:
    out_pdb_lines.append(F"REMARK {remark}" )

  __A		:     List[List[str]]
  if prot.parents is not None and len(prot.parents ) > 0:
    __A		:     Optional[int]												=		[]
    if prot.parents_chain_index is not None:
      __A		:     Dict[str, List[str]]												=		{}
      for p, i in zip(prot.parents			,     prot.parents_chain_index ):
        parent_dict.setdefault(str(__SCREAMING_SNAKE_CASE )			,     [] )
        parent_dict[str(__SCREAMING_SNAKE_CASE )].append(__SCREAMING_SNAKE_CASE )

      __A		:     Any												=		max([int(__SCREAMING_SNAKE_CASE ) for chain_idx in parent_dict] )
      for i in range(max_idx + 1 ):
        __A		:     Tuple												=		parent_dict.get(str(__SCREAMING_SNAKE_CASE )			,     ["""N/A"""] )
        parents_per_chain.append(__SCREAMING_SNAKE_CASE )
    else:
      parents_per_chain.append(list(prot.parents ) )
  else:
    __A		:     Dict												=		[["""N/A"""]]

  def make_parent_line(__SCREAMING_SNAKE_CASE						:  Sequence[str] ) -> str:
    return F"PARENT {' '.join(__SCREAMING_SNAKE_CASE )}"

  out_pdb_lines.append(make_parent_line(parents_per_chain[0] ) )

  __A		:     Optional[Any]												=		0
  for i, l in enumerate(__SCREAMING_SNAKE_CASE ):
    if "PARENT" not in l and "REMARK" not in l:
      out_pdb_lines.append(__SCREAMING_SNAKE_CASE )
    if "TER" in l and "END" not in lines[i + 1]:
      chain_counter += 1
      if not chain_counter >= len(__SCREAMING_SNAKE_CASE ):
        __A		:     Optional[Any]												=		parents_per_chain[chain_counter]
      else:
        __A		:     List[Any]												=		["""N/A"""]

      out_pdb_lines.append(make_parent_line(__SCREAMING_SNAKE_CASE ) )

  return "\n".join(__SCREAMING_SNAKE_CASE )







def 							A_     (						__SCREAMING_SNAKE_CASE						:  Protein )				->       str:





  """simple docstring"""

  __A		:     Optional[Any]												=		residue_constants.restypes + ["""X"""]

  def res_atoa(__SCREAMING_SNAKE_CASE						:  int ) -> str:
    return residue_constants.restype_atoa.get(restypes[r]			,     """UNK""" )

  __A		:     Any												=		residue_constants.atom_types

  __A		:     List[str]												=		[]

  __A		:     Union[str, Any]												=		prot.atom_mask
  __A		:     Optional[Any]												=		prot.aatype
  __A		:     List[Any]												=		prot.atom_positions
  __A		:     Union[str, Any]												=		prot.residue_index.astype(np.intaa )
  __A		:     Union[str, Any]												=		prot.b_factors
  __A		:     Union[str, Any]												=		prot.chain_index

  if np.any(aatype > residue_constants.restype_num ):
    raise ValueError("""Invalid aatypes.""" )

  __A		:     List[str]												=		get_pdb_headers(__SCREAMING_SNAKE_CASE )
  if len(__SCREAMING_SNAKE_CASE ) > 0:
    pdb_lines.extend(__SCREAMING_SNAKE_CASE )

  __A		:     Dict												=		aatype.shape[0]
  __A		:     int												=		1
  __A		:     Any												=		0
  __A		:     Optional[Any]												=		string.ascii_uppercase
  __A		:     int												=		None
  # Add all atom sites.
  for i in range(__SCREAMING_SNAKE_CASE ):
    __A		:     Optional[Any]												=		res_atoa(aatype[i] )
    for atom_name, pos, mask, b_factor in zip(__SCREAMING_SNAKE_CASE			,     atom_positions[i]			,     atom_mask[i]			,     b_factors[i] ):
      if mask < 0.5:
        continue

      __A		:     List[str]												=		"""ATOM"""
      __A		:     Tuple												=		atom_name if len(__SCREAMING_SNAKE_CASE ) == 4 else F" {atom_name}"
      __A		:     List[str]												=		""""""
      __A		:     Union[str, Any]												=		""""""
      __A		:     int												=		1.00
      __A		:     Any												=		atom_name[0]  # Protein supports only C, N, O, S, this works.
      __A		:     List[Any]												=		""""""

      __A		:     Union[str, Any]												=		"""A"""
      if chain_index is not None:
        __A		:     List[Any]												=		chain_tags[chain_index[i]]

      # PDB is a columnar format, every space matters here!
      __A		:     Tuple												=		(
          F"{record_type:<6}{atom_index:>5} {name:<4}{alt_loc:>1}"
          F"{res_name_a:>3} {chain_tag:>1}"
          F"{residue_index[i]:>4}{insertion_code:>1}   "
          F"{pos[0]:>8.3f}{pos[1]:>8.3f}{pos[2]:>8.3f}"
          F"{occupancy:>6.2f}{b_factor:>6.2f}          "
          F"{element:>2}{charge:>2}"
      )
      pdb_lines.append(__SCREAMING_SNAKE_CASE )
      atom_index += 1

    __A		:     List[str]												=		i == n - 1
    if chain_index is not None:
      if i != n - 1 and chain_index[i + 1] != prev_chain_index:
        __A		:     Dict												=		True
        __A		:     int												=		chain_index[i + 1]

    if should_terminate:
      # Close the chain.
      __A		:     List[Any]												=		"""TER"""
      __A		:     Optional[int]												=		(
          F"{chain_end:<6}{atom_index:>5}      {res_atoa(aatype[i] ):>3} {chain_tag:>1}{residue_index[i]:>4}"
      )
      pdb_lines.append(__SCREAMING_SNAKE_CASE )
      atom_index += 1

      if i != n - 1:
        # "prev" is a misnomer here. This happens at the beginning of
        # each new chain.
        pdb_lines.extend(get_pdb_headers(__SCREAMING_SNAKE_CASE			,     __SCREAMING_SNAKE_CASE ) )

  pdb_lines.append("""END""" )
  pdb_lines.append("""""" )
  return "\n".join(__SCREAMING_SNAKE_CASE )







def 							A_     (						__SCREAMING_SNAKE_CASE						:  Protein )				->       np.ndarray:





  """simple docstring"""

  return residue_constants.STANDARD_ATOM_MASK[prot.aatype]







def 							A_     (						__SCREAMING_SNAKE_CASE						:  FeatureDict			,     __SCREAMING_SNAKE_CASE						:  ModelOutput			,     __SCREAMING_SNAKE_CASE						:  Optional[np.ndarray] = None			,     __SCREAMING_SNAKE_CASE						:  Optional[np.ndarray] = None			,     __SCREAMING_SNAKE_CASE						:  Optional[str] = None			,     __SCREAMING_SNAKE_CASE						:  Optional[Sequence[str]] = None			,     __SCREAMING_SNAKE_CASE						:  Optional[Sequence[int]] = None			,     )				->       Protein:





  """simple docstring"""

  return Protein(
      aatype=features["""aatype"""]			,     atom_positions=result["""final_atom_positions"""]			,     atom_mask=result["""final_atom_mask"""]			,     residue_index=features["""residue_index"""] + 1			,     b_factors=b_factors if b_factors is not None else np.zeros_like(result["""final_atom_mask"""] )			,     chain_index=__SCREAMING_SNAKE_CASE			,     remark=__SCREAMING_SNAKE_CASE			,     parents=__SCREAMING_SNAKE_CASE			,     parents_chain_index=__SCREAMING_SNAKE_CASE			,     )





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



