





"""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	(			_lowercase		:					int      )				->Union[str, Any]:






	'''simple docstring'''






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

	return shapes






@torch.jit.ignore
def    _SCREAMING_SNAKE_CASE	(			_lowercase		:					int					, _lowercase		:					Dict      )				->Tuple:






	'''simple docstring'''






	a   :	List[Any]			=       []
	for d in reversed(__lowerCAmelCase      ):
		idx.append(flat_idx % d      )
		a   :	Dict			=       flat_idx // d

	return tuple(reversed(__lowerCAmelCase      )      )






@torch.jit.ignore
def    _SCREAMING_SNAKE_CASE	(			_lowercase		:					Optional[Any]					, _lowercase		:					str					, _lowercase		:					Optional[int]					, _lowercase		:					Union[str, Any] = None					, _lowercase		:					Optional[Any] = None					, )				->List[Any]:






	'''simple docstring'''






	def reduce_edge_list(_lowercase		:					Union[str, Any]      ) -> None:
		a   :	List[Any]			=       True
		for i in range(len(__lowerCAmelCase      )      ):
			a   :	Tuple			=       -1 * (i + 1)
			l[reversed_idx] &= tally
			a   :	str			=       l[reversed_idx]

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

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

	a   :	Tuple[slice, ...]			=       tuple(__lowerCAmelCase      )
	a   :	Tuple			=       len(__lowerCAmelCase      )

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

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

		a   :	List[str]			=       start[divergence_idx]
		return tuple(
		    path + (slice(__lowerCAmelCase					, 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   :	int			=       end[divergence_idx]
		return tuple(
		    path + (slice(__lowerCAmelCase					, 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   :	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    _SCREAMING_SNAKE_CASE	(			_lowercase		:					Optional[int]					, _lowercase		:					Dict					, _lowercase		:					str					, _lowercase		:					Union[str, Any]      )				->Union[str, Any]:






	'''simple docstring'''






	a   :	Any			=       t.shape[:no_batch_dims]
	a   :	Dict			=       list(_flat_idx_to_idx(__lowerCAmelCase					, __lowerCAmelCase      )      )
	# _get_minimal_slice_set is inclusive
	a   :	Optional[int]			=       list(_flat_idx_to_idx(flat_end - 1					, __lowerCAmelCase      )      )

	# Get an ordered list of slices to perform
	a   :	Union[str, Any]			=       _get_minimal_slice_set(
	    __lowerCAmelCase					, __lowerCAmelCase					, __lowerCAmelCase					, )

	a   :	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    _SCREAMING_SNAKE_CASE	(			_lowercase		:					Optional[Any]					, _lowercase		:					int					, _lowercase		:					str					, _lowercase		:					List[str]					, _lowercase		:					Tuple = False					, _lowercase		:					int = None					, _lowercase		:					Tuple = False					, )				->str:






	'''simple docstring'''






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

	a   :	int			=       [shape[:no_batch_dims] for shape in _fetch_dims(__lowerCAmelCase      )]
	a   :	Optional[int]			=       tuple([max(__lowerCAmelCase      ) for s in zip(*__lowerCAmelCase      )]      )

	def _prep_inputs(_lowercase		:					str      ) -> 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   :	str			=       t.reshape(-1					, *t.shape[no_batch_dims:]      )
		else:
			a   :	List[str]			=       t.expand(orig_batch_dims + t.shape[no_batch_dims:]      )
		return t

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

	a   :	List[Any]			=       1
	for d in orig_batch_dims:
		flat_batch_dim *= d

	a   :	List[Any]			=       flat_batch_dim // chunk_size + (flat_batch_dim % chunk_size != 0)

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

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

		a   :	Dict[str, Any]			=       tensor_tree_map(__lowerCAmelCase					, __lowerCAmelCase      )

		# Run the layer on the chunk
		a   :	Optional[int]			=       layer(**__lowerCAmelCase      )

		# Allocate space for the output
		if out is None:
			a   :	Optional[Any]			=       tensor_tree_map(lambda _lowercase      : t.new_zeros((flat_batch_dim,) + t.shape[1:]      )					, __lowerCAmelCase      )

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

			def assign(_lowercase		:					Union[str, Any]					, _lowercase		:					List[Any]      ) -> None:
				for k, v in da.items():
					if isinstance(__lowerCAmelCase					, __lowerCAmelCase      ):
						assign(__lowerCAmelCase					, da[k]      )
					else:
						if _add_into_out:
							v[i : i + chunk_size] += da[k]
						else:
							a   :	str			=       da[k]

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

	return out







class  __UpperCamelCase       :






							def __init__(     self   ,      lowerCAmelCase__ = 512   ,      )			->				str:
								a   :	str			=       max_chunk_size
								a   :	Optional[int]			=       None
								a   :	Optional[tuple]			=       None






							def 					__a     (     self   ,      lowerCAmelCase__   ,      lowerCAmelCase__   ,      lowerCAmelCase__   )			->				int:
								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   :	int			=       [min_chunk_size] + candidates
								candidates[-1] += 4

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

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

								return candidates[min_viable_chunk_size_index]






							def 					__a     (     self   ,      lowerCAmelCase__   ,      lowerCAmelCase__   )			->				bool:
								a   :	List[str]			=       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   :	str			=       [v for _, v in sorted(aa.items()   ,      key=lambda lowerCAmelCase__   : x[0]   )]
										a   :	List[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 					__a     (     self   ,      lowerCAmelCase__   ,      lowerCAmelCase__   ,      lowerCAmelCase__   ,      )			->				int:
								a   :	Optional[Any]			=       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   :	Dict			=       self._compare_arg_caches(self.cached_arg_data   ,      lowerCamelCase__   )
								else:
									# Otherwise, we can reuse the precomputed value
									a   :	int			=       False

								if not consistent:
									a   :	Any			=       self._determine_favorable_chunk_size(
									    lowerCamelCase__   ,      lowerCamelCase__   ,      lowerCamelCase__   ,      )
									a   :	str			=       arg_data

								assert self.cached_chunk_size is not None

								return self.cached_chunk_size









"""simple docstring"""




import baseaa






def    _SCREAMING_SNAKE_CASE	(			_lowercase		:					str      )				->bytes:






	'''simple docstring'''






	return baseaa.aaaencode(string.encode("utf-8"      )      )






def    _SCREAMING_SNAKE_CASE	(			_lowercase		:					bytes      )				->str:






	'''simple docstring'''






	return baseaa.aaadecode(_lowercase      ).decode("utf-8"      )


if __name__ == "__main__":
		import doctest

		doctest.testmod()


