





'''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 						__lowerCamelCase			(	UpperCAmelCase_							)   ->List[Tuple[int, ...]]:
	snake_case__	       =   []
	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 						__lowerCamelCase			(	UpperCAmelCase_					,				UpperCAmelCase_							)   ->Tuple[int, ...]:
	snake_case__	       =   []
	for d in reversed(UpperCAmelCase_							):
		idx.append(flat_idx % d							)
		snake_case__	       =   flat_idx // d

	return tuple(reversed(UpperCAmelCase_							)							)




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

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

	if start_edges is None:
		snake_case__	       =   [s == 0 for s in start]
		reduce_edge_list(UpperCAmelCase_							)
	if end_edges is None:
		snake_case__	       =   [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							),)]

	snake_case__	       =   []
	snake_case__	       =   []

	# 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

	snake_case__	       =   tuple(UpperCAmelCase_							)
	snake_case__	       =   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

		snake_case__	       =   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

		snake_case__	       =   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()							)
		snake_case__	       =   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 						__lowerCamelCase			(	UpperCAmelCase_					,				UpperCAmelCase_					,				UpperCAmelCase_					,				UpperCAmelCase_							)   ->torch.Tensor:
	snake_case__	       =   t.shape[:no_batch_dims]
	snake_case__	       =   list(_flat_idx_to_idx(UpperCAmelCase_					,				UpperCAmelCase_							)							)
	# _get_minimal_slice_set is inclusive
	snake_case__	       =   list(_flat_idx_to_idx(flat_end - 1					,				UpperCAmelCase_							)							)

	# Get an ordered list of slices to perform
	snake_case__	       =   _get_minimal_slice_set(
	    UpperCAmelCase_					,				UpperCAmelCase_					,				UpperCAmelCase_					,				)

	snake_case__	       =   [t[s] for s in slices]

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




def 						__lowerCamelCase			(	UpperCAmelCase_					,				UpperCAmelCase_					,				UpperCAmelCase_					,				UpperCAmelCase_					,				UpperCAmelCase_ = False					,				UpperCAmelCase_ = None					,				UpperCAmelCase_ = False					,				)   ->Any:
	if not (len(UpperCAmelCase_							) > 0):
		raise ValueError('Must provide at least one input'							)

	snake_case__	       =   [shape[:no_batch_dims] for shape in _fetch_dims(UpperCAmelCase_							)]
	snake_case__	       =   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:
				snake_case__	       =   t.expand(orig_batch_dims + t.shape[no_batch_dims:]							)
			snake_case__	       =   t.reshape(-1					,				*t.shape[no_batch_dims:]							)
		else:
			snake_case__	       =   t.expand(orig_batch_dims + t.shape[no_batch_dims:]							)
		return t

	snake_case__	       =   tensor_tree_map(_prep_inputs					,				UpperCAmelCase_							)
	snake_case__	       =   None
	if _out is not None:
		snake_case__	       =   tensor_tree_map(lambda UpperCAmelCase_							: t.view([-1] + list(t.shape[no_batch_dims:]							)							)					,				_out							)

	snake_case__	       =   1
	for d in orig_batch_dims:
		flat_batch_dim *= d

	snake_case__	       =   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

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

		snake_case__	       =   tensor_tree_map(UpperCAmelCase_					,				UpperCAmelCase_							)

		# Run the layer on the chunk
		snake_case__	       =   layer(**UpperCAmelCase_							)

		# Allocate space for the output
		if out is None:
			snake_case__	       =   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:
							snake_case__	       =   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:
					snake_case__	       =   xa
		elif isinstance(UpperCAmelCase_					,				torch.Tensor							):
			if _add_into_out:
				out[i : i + chunk_size] += output_chunk
			else:
				snake_case__	       =   output_chunk
		else:
			raise ValueError('Not supported'							)

		i += chunk_size

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

	return out







class 				__snake_case     :





	def __init__(		self      ,       UpperCamelCase_ = 512      ,       )      ->		Tuple:
		snake_case__	       =   max_chunk_size
		snake_case__	       =   None
		snake_case__	       =   None





	def   _snake_case   (		self      ,       UpperCamelCase_      ,       UpperCamelCase_      ,       UpperCamelCase_		)      ->		int:
		logging.info('Tuning chunk size...'		)

		if min_chunk_size >= self.max_chunk_size:
			return min_chunk_size

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

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

		snake_case__	       =   0
		snake_case__	       =   len(UpperCamelCase_		) - 1
		while i > min_viable_chunk_size_index:
			snake_case__	       =   test_chunk_size(candidates[i]		)
			if not viable:
				snake_case__	       =   (min_viable_chunk_size_index + i) // 2
			else:
				snake_case__	       =   i
				snake_case__	       =   (i + len(UpperCamelCase_		) - 1) // 2

		return candidates[min_viable_chunk_size_index]





	def   _snake_case   (		self      ,       UpperCamelCase_      ,       UpperCamelCase_		)      ->		bool:
		snake_case__	       =   True
		for aa, aa in zip(UpperCamelCase_      ,       UpperCamelCase_		):
			assert type(UpperCamelCase_		) == type(UpperCamelCase_		)
			if isinstance(UpperCamelCase_      ,       (list, tuple)		):
				consistent &= self._compare_arg_caches(UpperCamelCase_      ,       UpperCamelCase_		)
			elif isinstance(UpperCamelCase_      ,       UpperCamelCase_		):
				snake_case__	       =   [v for _, v in sorted(aa.items()      ,       key=lambda UpperCamelCase_		: x[0]		)]
				snake_case__	       =   [v for _, v in sorted(aa.items()      ,       key=lambda UpperCamelCase_		: x[0]		)]
				consistent &= self._compare_arg_caches(UpperCamelCase_      ,       UpperCamelCase_		)
			else:
				consistent &= aa == aa

		return consistent





	def   _snake_case   (		self      ,       UpperCamelCase_      ,       UpperCamelCase_      ,       UpperCamelCase_      ,       )      ->		int:
		snake_case__	       =   True
		snake_case__	       =   tree_map(lambda UpperCamelCase_		: a.shape if isinstance(UpperCamelCase_      ,       torch.Tensor		) else a      ,       UpperCamelCase_      ,       UpperCamelCase_		)
		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(UpperCamelCase_		)
			snake_case__	       =   self._compare_arg_caches(self.cached_arg_data      ,       UpperCamelCase_		)
		else:
			# Otherwise, we can reuse the precomputed value
			snake_case__	       =   False

		if not consistent:
			snake_case__	       =   self._determine_favorable_chunk_size(
			    UpperCamelCase_      ,       UpperCamelCase_      ,       UpperCamelCase_      ,       )
			snake_case__	       =   arg_data

		assert self.cached_chunk_size is not None

		return self.cached_chunk_size










'''simple docstring'''




from math import loga




def 						__lowerCamelCase			(	UpperCAmelCase_							)   ->int:
	if a < 0:
		raise ValueError('Input value must be a positive integer'							)
	elif isinstance(UpperCAmelCase_					,				UpperCAmelCase_							):
		raise TypeError('Input value must be a \'int\' type'							)
	return 0 if (a == 0) else int(loga(a & -a							)							)


if __name__ == "__main__":
			import doctest

			doctest.testmod()



