






'''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 			SCREAMING_SNAKE_CASE_	(UpperCamelCase				)				->	List[Tuple[int, ...]]:
							lowerCamelCase__ :     List[str]	  =	[]
							if isinstance(UpperCamelCase    ,      UpperCamelCase				):
														for v in tree.values():
																					shapes.extend(_fetch_dims(UpperCamelCase				)				)
							elif isinstance(UpperCamelCase    ,      (list, tuple)				):
														for t in tree:
																					shapes.extend(_fetch_dims(UpperCamelCase				)				)
							elif isinstance(UpperCamelCase    ,      torch.Tensor				):
														shapes.append(tree.shape				)
							else:
														raise ValueError("""Not supported"""				)

							return shapes





@torch.jit.ignore
def 			SCREAMING_SNAKE_CASE_	(UpperCamelCase    ,      UpperCamelCase				)				->	Tuple[int, ...]:
							lowerCamelCase__ :     List[Any]	  =	[]
							for d in reversed(UpperCamelCase				):
														idx.append(flat_idx % d				)
														lowerCamelCase__ :     Dict	  =	flat_idx // d

							return tuple(reversed(UpperCamelCase				)				)





@torch.jit.ignore
def 			SCREAMING_SNAKE_CASE_	(UpperCamelCase    ,      UpperCamelCase    ,      UpperCamelCase    ,      UpperCamelCase = None    ,      UpperCamelCase = None    ,      )				->	List[Tuple[slice, ...]]:



							def reduce_edge_list(UpperCamelCase				) -> None:
														lowerCamelCase__ :     int	  =	True
														for i in range(len(UpperCamelCase				)				):
																					lowerCamelCase__ :     List[Any]	  =	-1 * (i + 1)
																					l[reversed_idx] &= tally
																					lowerCamelCase__ :     List[str]	  =	l[reversed_idx]

							if start_edges is None:
														lowerCamelCase__ :     Dict	  =	[s == 0 for s in start]
														reduce_edge_list(UpperCamelCase				)
							if end_edges is None:
														lowerCamelCase__ :     Optional[int]	  =	[e == (d - 1) for e, d in zip(UpperCamelCase    ,      UpperCamelCase				)]
														reduce_edge_list(UpperCamelCase				)

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

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

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

							lowerCamelCase__ :     Tuple[slice, ...]	  =	tuple(UpperCamelCase				)
							lowerCamelCase__ :     int	  =	len(UpperCamelCase				)

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

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

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

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

							# Get an ordered list of slices to perform
							lowerCamelCase__ :     int	  =	_get_minimal_slice_set(
							    UpperCamelCase    ,      UpperCamelCase    ,      UpperCamelCase    ,      )

							lowerCamelCase__ :     Any	  =	[t[s] for s in slices]

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





def 			SCREAMING_SNAKE_CASE_	(UpperCamelCase    ,      UpperCamelCase    ,      UpperCamelCase    ,      UpperCamelCase    ,      UpperCamelCase = False    ,      UpperCamelCase = None    ,      UpperCamelCase = False    ,      )				->	Any:



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

							lowerCamelCase__ :     str	  =	[shape[:no_batch_dims] for shape in _fetch_dims(UpperCamelCase				)]
							lowerCamelCase__ :     Optional[int]	  =	tuple([max(UpperCamelCase				) for s in zip(*UpperCamelCase				)]				)

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

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

							lowerCamelCase__ :     Dict	  =	1
							for d in orig_batch_dims:
														flat_batch_dim *= d

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

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

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

														lowerCamelCase__ :     Dict[str, Any]	  =	tensor_tree_map(UpperCamelCase    ,      UpperCamelCase				)

														# Run the layer on the chunk
														lowerCamelCase__ :     List[Any]	  =	layer(**UpperCamelCase				)

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

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

																					def assign(UpperCamelCase    ,      UpperCamelCase				) -> None:
																												for k, v in da.items():
																																			if isinstance(UpperCamelCase    ,      UpperCamelCase				):
																																										assign(UpperCamelCase    ,      da[k]				)
																																			else:
																																										if _add_into_out:
																																																	v[i : i + chunk_size] += da[k]
																																										else:
																																																	lowerCamelCase__ :     Tuple	  =	da[k]

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

														i += chunk_size

							lowerCamelCase__ :     List[Any]	  =	tensor_tree_map(lambda UpperCamelCase				: t.view(orig_batch_dims + t.shape[1:]				)    ,      UpperCamelCase				)

							return out



class _lowercase     :




							def __init__(							self:    str      ,							UpperCamelCase__:    int = 512      ,							):
														lowerCamelCase__ :     Any	  =	max_chunk_size
														lowerCamelCase__ :     Optional[int]	  =	None
														lowerCamelCase__ :     Optional[tuple]	  =	None




							def 		lowerCamelCase_    (							self:    Any      ,							UpperCamelCase__:    Callable      ,							UpperCamelCase__:    tuple      ,							UpperCamelCase__:    int					):
														logging.info("""Tuning chunk size..."""					)

														if min_chunk_size >= self.max_chunk_size:
																					return min_chunk_size

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

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

														lowerCamelCase__ :     Dict	  =	0
														lowerCamelCase__ :     Dict	  =	len(__lowerCamelCase					) - 1
														while i > min_viable_chunk_size_index:
																					lowerCamelCase__ :     str	  =	test_chunk_size(candidates[i]					)
																					if not viable:
																												lowerCamelCase__ :     List[str]	  =	(min_viable_chunk_size_index + i) // 2
																					else:
																												lowerCamelCase__ :     List[Any]	  =	i
																												lowerCamelCase__ :     Dict	  =	(i + len(__lowerCamelCase					) - 1) // 2

														return candidates[min_viable_chunk_size_index]




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

														return consistent




							def 		lowerCamelCase_    (							self:    List[Any]      ,							UpperCamelCase__:    Callable      ,							UpperCamelCase__:    tuple      ,							UpperCamelCase__:    int      ,							):
														lowerCamelCase__ :     Optional[Any]	  =	True
														lowerCamelCase__ :     tuple	  =	tree_map(lambda UpperCamelCase__					: 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					)
																					lowerCamelCase__ :     Union[str, Any]	  =	self._compare_arg_caches(self.cached_arg_data      ,							__lowerCamelCase					)
														else:
																					# Otherwise, we can reuse the precomputed value
																					lowerCamelCase__ :     Any	  =	False

														if not consistent:
																					lowerCamelCase__ :     List[str]	  =	self._determine_favorable_chunk_size(
																					    __lowerCamelCase      ,							__lowerCamelCase      ,							__lowerCamelCase      ,							)
																					lowerCamelCase__ :     Dict	  =	arg_data

														assert self.cached_chunk_size is not None

														return self.cached_chunk_size













'''simple docstring'''







from torch import nn





def 			SCREAMING_SNAKE_CASE_	(UpperCamelCase				)				->	Dict:
							if act_fn in ["swish", "silu"]:
														return nn.SiLU()
							elif act_fn == "mish":
														return nn.Mish()
							elif act_fn == "gelu":
														return nn.GELU()
							else:
														raise ValueError(f'''Unsupported activation function: {act_fn}'''				)

