



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       (      _lowerCamelCase						):





							'''simple docstring'''




							_lowerCAmelCase  :							Union[str, 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        A       (      _lowerCamelCase       ,				_lowerCamelCase						):





							'''simple docstring'''




							_lowerCAmelCase  :							Tuple				     =					[]
							for d in reversed(_lowerCamelCase						):
														idx.append(flat_idx % d						)
														_lowerCAmelCase  :							Any				     =					flat_idx // d

							return tuple(reversed(_lowerCamelCase						)						)






@torch.jit.ignore
def        A       (      _lowerCamelCase       ,				_lowerCamelCase       ,				_lowerCamelCase       ,				_lowerCamelCase = None       ,				_lowerCamelCase = None       ,				):





							'''simple docstring'''




							def reduce_edge_list(_lowerCamelCase						) -> None:
														_lowerCAmelCase  :							Dict				     =					True
														for i in range(len(_lowerCamelCase						)						):
																					_lowerCAmelCase  :							Tuple				     =					-1 * (i + 1)
																					l[reversed_idx] &= tally
																					_lowerCAmelCase  :							List[str]				     =					l[reversed_idx]

							if start_edges is None:
														_lowerCAmelCase  :							List[Any]				     =					[s == 0 for s in start]
														reduce_edge_list(_lowerCamelCase						)
							if end_edges is None:
														_lowerCAmelCase  :							Dict				     =					[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						),)]

							_lowerCAmelCase  :							List[Tuple[slice, ...]]				     =					[]
							_lowerCAmelCase  :							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

							_lowerCAmelCase  :							Tuple[slice, ...]				     =					tuple(_lowerCamelCase						)
							_lowerCAmelCase  :							str				     =					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

														_lowerCAmelCase  :							List[Any]				     =					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

														_lowerCAmelCase  :							Tuple				     =					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()						)
														_lowerCAmelCase  :							Dict				     =					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       (      _lowerCamelCase       ,				_lowerCamelCase       ,				_lowerCamelCase       ,				_lowerCamelCase						):





							'''simple docstring'''




							_lowerCAmelCase  :							Optional[int]				     =					t.shape[:no_batch_dims]
							_lowerCAmelCase  :							int				     =					list(_flat_idx_to_idx(_lowerCamelCase       ,				_lowerCamelCase						)						)
							# _get_minimal_slice_set is inclusive
							_lowerCAmelCase  :							Union[str, Any]				     =					list(_flat_idx_to_idx(flat_end - 1       ,				_lowerCamelCase						)						)

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

							_lowerCAmelCase  :							Dict				     =					[t[s] for s in slices]

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






def        A       (      _lowerCamelCase       ,				_lowerCamelCase       ,				_lowerCamelCase       ,				_lowerCamelCase       ,				_lowerCamelCase = False       ,				_lowerCamelCase = None       ,				_lowerCamelCase = False       ,				):





							'''simple docstring'''




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

							_lowerCAmelCase  :							Tuple				     =					[shape[:no_batch_dims] for shape in _fetch_dims(_lowerCamelCase						)]
							_lowerCAmelCase  :							Union[str, Any]				     =					tuple([max(_lowerCamelCase						) for s in zip(*_lowerCamelCase						)]						)

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

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

							_lowerCAmelCase  :							Union[str, Any]				     =					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(_lowerCamelCase						) -> torch.Tensor:
														return t[i : i + chunk_size] if t.shape[0] != 1 else t

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

														_lowerCAmelCase  :							Dict[str, Any]				     =					tensor_tree_map(_lowerCamelCase       ,				_lowerCamelCase						)

														# Run the layer on the chunk
														_lowerCAmelCase  :							Union[str, Any]				     =					layer(**_lowerCamelCase						)

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

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

																					def assign(_lowerCamelCase       ,				_lowerCamelCase						) -> 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:
																																																	_lowerCAmelCase  :							Any				     =					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:
																																			_lowerCAmelCase  :							Dict				     =					xa
														elif isinstance(_lowerCamelCase       ,				torch.Tensor						):
																					if _add_into_out:
																												out[i : i + chunk_size] += output_chunk
																					else:
																												_lowerCAmelCase  :							List[Any]				     =					output_chunk
														else:
																					raise ValueError("Not supported"						)

														i += chunk_size

							_lowerCAmelCase  :							Dict				     =					tensor_tree_map(lambda _lowerCamelCase						: t.view(orig_batch_dims + t.shape[1:]						)       ,				_lowerCamelCase						)

							return out



class 						UpperCAmelCase_					:







							def __init__(     self,    __a = 512,    ):

														'''simple docstring'''

														_lowerCAmelCase  :							List[str]				     =					max_chunk_size
														_lowerCAmelCase  :							Optional[int]				     =					None
														_lowerCAmelCase  :							Optional[tuple]				     =					None







							def      snake_case__   (     self,    __a,    __a,    __a):

														'''simple docstring'''

														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  :							Dict				     =					[c for c in candidates if c > min_chunk_size]
														_lowerCAmelCase  :							Optional[int]				     =					[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

														_lowerCAmelCase  :							Any				     =					0
														_lowerCAmelCase  :							Tuple				     =					len(__a) - 1
														while i > min_viable_chunk_size_index:
																					_lowerCAmelCase  :							int				     =					test_chunk_size(candidates[i])
																					if not viable:
																												_lowerCAmelCase  :							Tuple				     =					(min_viable_chunk_size_index + i) // 2
																					else:
																												_lowerCAmelCase  :							Optional[Any]				     =					i
																												_lowerCAmelCase  :							Optional[Any]				     =					(i + len(__a) - 1) // 2

														return candidates[min_viable_chunk_size_index]







							def      snake_case__   (     self,    __a,    __a):

														'''simple docstring'''

														_lowerCAmelCase  :							str				     =					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):
																												_lowerCAmelCase  :							Any				     =					[v for _, v in sorted(aa.items(),    key=lambda __a: x[0])]
																												_lowerCAmelCase  :							Optional[Any]				     =					[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,    ):

														'''simple docstring'''

														_lowerCAmelCase  :							Tuple				     =					True
														_lowerCAmelCase  :							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)
																					_lowerCAmelCase  :							Dict				     =					self._compare_arg_caches(self.cached_arg_data,    __a)
														else:
																					# Otherwise, we can reuse the precomputed value
																					_lowerCAmelCase  :							Dict				     =					False

														if not consistent:
																					_lowerCAmelCase  :							Union[str, Any]				     =					self._determine_favorable_chunk_size(
																					    __a,    __a,    __a,    )
																					_lowerCAmelCase  :							Optional[Any]				     =					arg_data

														assert self.cached_chunk_size is not None

														return self.cached_chunk_size





import baseaa






def        A       (      _lowerCamelCase						):





							'''simple docstring'''




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






def        A       (      _lowerCamelCase						):





							'''simple docstring'''




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


if __name__ == "__main__":
						import doctest

						doctest.testmod()
