






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 					snake_case_						(   lowerCAmelCase_				:  Union[dict, list, tuple, torch.Tensor]					):
							__lowercase					:							List[str]			=   []
							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 					snake_case_						(   lowerCAmelCase_				:  int				,				lowerCAmelCase_				:  Tuple[int, ...]					):
							__lowercase					:							Union[str, Any]			=   []
							for d in reversed(lowerCAmelCase_					):
														idx.append(flat_idx % d					)
														__lowercase					:							Dict			=   flat_idx // d

							return tuple(reversed(lowerCAmelCase_					)					)
@torch.jit.ignore
def 					snake_case_						(   lowerCAmelCase_				:  Sequence[int]				,				lowerCAmelCase_				:  Sequence[int]				,				lowerCAmelCase_				:  Sequence[int]				,				lowerCAmelCase_				:  Optional[Sequence[bool]] = None				,				lowerCAmelCase_				:  Optional[Sequence[bool]] = None				,				):

							# 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(lowerCAmelCase_				:  List[bool]					) -> None:
														__lowercase					:							Tuple			=   True
														for i in range(len(lowerCAmelCase_					)					):
																					__lowercase					:							List[str]			=   -1 * (i + 1)
																					l[reversed_idx] &= tally
																					__lowercase					:							List[str]			=   l[reversed_idx]

							if start_edges is None:
														__lowercase					:							Union[str, Any]			=   [s == 0 for s in start]
														reduce_edge_list(lowerCAmelCase_					)
							if end_edges is None:
														__lowercase					:							Any			=   [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					),)]

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

							__lowercase					:							Tuple[slice, ...]			=   tuple(lowerCAmelCase_					)
							__lowercase					:							Any			=   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

														__lowercase					:							Tuple			=   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

														__lowercase					:							Dict			=   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()					)
														__lowercase					:							Optional[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 					snake_case_						(   lowerCAmelCase_				:  torch.Tensor				,				lowerCAmelCase_				:  int				,				lowerCAmelCase_				:  int				,				lowerCAmelCase_				:  int					):
							__lowercase					:							Union[str, Any]			=   t.shape[:no_batch_dims]
							__lowercase					:							Optional[int]			=   list(_flat_idx_to_idx(lowerCAmelCase_				,				lowerCAmelCase_					)					)
							# _get_minimal_slice_set is inclusive
							__lowercase					:							Optional[Any]			=   list(_flat_idx_to_idx(flat_end - 1				,				lowerCAmelCase_					)					)

							# Get an ordered list of slices to perform
							__lowercase					:							Any			=   _get_minimal_slice_set(
							    lowerCAmelCase_				,				lowerCAmelCase_				,				lowerCAmelCase_				,				)

							__lowercase					:							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 					snake_case_						(   lowerCAmelCase_				:  Callable				,				lowerCAmelCase_				:  Dict[str, Any]				,				lowerCAmelCase_				:  int				,				lowerCAmelCase_				:  int				,				lowerCAmelCase_				:  bool = False				,				lowerCAmelCase_				:  Any = None				,				lowerCAmelCase_				:  bool = False				,				):
							if not (len(lowerCAmelCase_					) > 0):
														raise ValueError("""Must provide at least one input"""					)

							__lowercase					:							Union[str, Any]			=   [shape[:no_batch_dims] for shape in _fetch_dims(lowerCAmelCase_					)]
							__lowercase					:							int			=   tuple([max(lowerCAmelCase_					) for s in zip(*lowerCAmelCase_					)]					)

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

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

							__lowercase					:							Union[str, Any]			=   1
							for d in orig_batch_dims:
														flat_batch_dim *= d

							__lowercase					:							int			=   flat_batch_dim // chunk_size + (flat_batch_dim % chunk_size != 0)

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

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

														__lowercase					:							Dict[str, Any]			=   tensor_tree_map(lowerCAmelCase_				,				lowerCAmelCase_					)

														# Run the layer on the chunk
														__lowercase					:							Tuple			=   layer(**lowerCAmelCase_					)

														# Allocate space for the output
														if out is None:
																					__lowercase					:							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_				:  dict				,				lowerCAmelCase_				:  dict					) -> 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:
																																																	__lowercase					:							List[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:
																																			__lowercase					:							Tuple			=   xa
														elif isinstance(lowerCAmelCase_				,				torch.Tensor					):
																					if _add_into_out:
																												out[i : i + chunk_size] += output_chunk
																					else:
																												__lowercase					:							List[str]			=   output_chunk
														else:
																					raise ValueError("""Not supported"""					)

														i += chunk_size

							__lowercase					:							Any			=   tensor_tree_map(lambda lowerCAmelCase_					: t.view(orig_batch_dims + t.shape[1:]					)				,				lowerCAmelCase_					)

							return out



class 						lowerCAmelCase   :

					'''simple docstring'''

					def __init__(     self :		int						,   __a :		int = 512						,   )	->     Any:


												"""simple docstring"""

												__lowercase					:							Union[str, Any]			=   max_chunk_size
												__lowercase					:							Optional[int]			=   None
												__lowercase					:							Optional[tuple]			=   None



					def 					lowerCAmelCase (     self :		Optional[Any]						,   __a :		Callable						,   __a :		tuple						,   __a :		int							)	->     int:


												"""simple docstring"""

												logging.info("""Tuning chunk size..."""							)

												if min_chunk_size >= self.max_chunk_size:
																			return min_chunk_size

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

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

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

												return candidates[min_viable_chunk_size_index]



					def 					lowerCAmelCase (     self :		Optional[Any]						,   __a :		Iterable						,   __a :		Iterable							)	->     bool:


												"""simple docstring"""

												__lowercase					:							Any			=   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							):
																										__lowercase					:							str			=   [v for _, v in sorted(aa.items()						,   key=lambda __a							: x[0]							)]
																										__lowercase					:							Optional[int]			=   [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 					lowerCAmelCase (     self :		Union[str, Any]						,   __a :		Callable						,   __a :		tuple						,   __a :		int						,   )	->     int:


												"""simple docstring"""

												__lowercase					:							int			=   True
												__lowercase					:							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							)
																			__lowercase					:							str			=   self._compare_arg_caches(self.cached_arg_data						,   __a							)
												else:
																			# Otherwise, we can reuse the precomputed value
																			__lowercase					:							List[Any]			=   False

												if not consistent:
																			__lowercase					:							int			=   self._determine_favorable_chunk_size(
																			    __a						,   __a						,   __a						,   )
																			__lowercase					:							Dict			=   arg_data

												assert self.cached_chunk_size is not None

												return self.cached_chunk_size







import math
def 					snake_case_						(   lowerCAmelCase_				:  int					):

							if 1 < number < 4:
														# 2 and 3 are primes
														return True
							elif number < 2 or number % 2 == 0 or number % 3 == 0:
														# Negatives, 0, 1, all even numbers, all multiples of 3 are not primes
														return False

							# All primes number are in format of 6k +/- 1
							for i in range(5				,				int(math.sqrt(lowerCAmelCase_					) + 1					)				,				6					):
														if number % i == 0 or number % (i + 2) == 0:
																					return False
							return True
def 					snake_case_						(   lowerCAmelCase_				:  int = 10001					):

							try:
														__lowercase					:							Optional[int]			=   int(lowerCAmelCase_					)
							except (TypeError, ValueError):
														raise TypeError("""Parameter nth must be int or castable to int."""					) from None
							if nth <= 0:
														raise ValueError("""Parameter nth must be greater than or equal to one."""					)
							__lowercase					:							list[int]			=   []
							__lowercase					:							str			=   2
							while len(lowerCAmelCase_					) < nth:
														if is_prime(lowerCAmelCase_					):
																					primes.append(lowerCAmelCase_					)
																					num += 1
														else:
																					num += 1
							return primes[len(lowerCAmelCase_					) - 1]


if __name__ == "__main__":
	print(f'''{solution() = }''')