


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							(     _lowerCAmelCase			)    ->						List[Tuple[int, ...]]:


						'''simple docstring'''






						__snake_case				=		[]
						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 						_lowerCAmelCase							(     _lowerCAmelCase   ,						_lowerCAmelCase			)    ->						Tuple[int, ...]:


						'''simple docstring'''






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

						return tuple(reversed(_lowerCAmelCase			)			)





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


						'''simple docstring'''






						def reduce_edge_list(_lowerCAmelCase			) -> None:
												__snake_case				=		True
												for i in range(len(_lowerCAmelCase			)			):
																		__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(_lowerCAmelCase			)
						if end_edges is None:
												__snake_case				=		[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			),)]

						__snake_case				=		[]
						__snake_case				=		[]

						# 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

						__snake_case				=		tuple(_lowerCAmelCase			)
						__snake_case				=		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

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

												__snake_case				=		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()			)
												__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							(     _lowerCAmelCase   ,						_lowerCAmelCase   ,						_lowerCAmelCase   ,						_lowerCAmelCase			)    ->						torch.Tensor:


						'''simple docstring'''






						__snake_case				=		t.shape[:no_batch_dims]
						__snake_case				=		list(_flat_idx_to_idx(_lowerCAmelCase   ,						_lowerCAmelCase			)			)
						# _get_minimal_slice_set is inclusive
						__snake_case				=		list(_flat_idx_to_idx(flat_end - 1   ,						_lowerCAmelCase			)			)

						# Get an ordered list of slices to perform
						__snake_case				=		_get_minimal_slice_set(
						    _lowerCAmelCase   ,						_lowerCAmelCase   ,						_lowerCAmelCase   ,						)

						__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							(     _lowerCAmelCase   ,						_lowerCAmelCase   ,						_lowerCAmelCase   ,						_lowerCAmelCase   ,						_lowerCAmelCase = False   ,						_lowerCAmelCase = None   ,						_lowerCAmelCase = False   ,						)    ->						Any:


						'''simple docstring'''






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

						__snake_case				=		[shape[:no_batch_dims] for shape in _fetch_dims(_lowerCAmelCase			)]
						__snake_case				=		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:
																								__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   ,						_lowerCAmelCase			)
						__snake_case				=		None
						if _out is not None:
												__snake_case				=		tensor_tree_map(lambda _lowerCAmelCase			: 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(_lowerCAmelCase			) -> 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(_lowerCAmelCase			):
												# Chunk the input
												if not low_mem:
																		__snake_case				=		_select_chunk
												else:
																		__snake_case				=		partial(
																		    _chunk_slice   ,						flat_start=_lowerCAmelCase   ,						flat_end=min(_lowerCAmelCase   ,						i + chunk_size			)   ,						no_batch_dims=len(_lowerCAmelCase			)   ,						)

												__snake_case				=		tensor_tree_map(_lowerCAmelCase   ,						_lowerCAmelCase			)

												# Run the layer on the chunk
												__snake_case				=		layer(**_lowerCAmelCase			)

												# Allocate space for the output
												if out is None:
																		__snake_case				=		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:
																																										__snake_case				=		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:
																														__snake_case				=		xa
												elif isinstance(_lowerCAmelCase   ,						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 _lowerCAmelCase			: t.view(orig_batch_dims + t.shape[1:]			)   ,						_lowerCAmelCase			)

						return out






class      UpperCamelCase:


					def __init__(    self						:		int  ,     SCREAMING_SNAKE_CASE						:		int = 5_1_2  ,     )      ->						int:




											'''simple docstring'''

											__snake_case				=		max_chunk_size
											__snake_case				=		None
											__snake_case				=		None


					def 				SCREAMING_SNAKE_CASE_	(    self						:		Dict  ,     SCREAMING_SNAKE_CASE						:		Callable  ,     SCREAMING_SNAKE_CASE						:		tuple  ,     SCREAMING_SNAKE_CASE						:		int    )      ->						int:




											'''simple docstring'''

											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(SCREAMING_SNAKE_CASE						:		int    ) -> bool:
																	try:
																							with torch.no_grad():
																													fn(*SCREAMING_SNAKE_CASE  ,     chunk_size=SCREAMING_SNAKE_CASE    )
																							return True
																	except RuntimeError:
																							return False

											__snake_case				=		0
											__snake_case				=		len(SCREAMING_SNAKE_CASE    ) - 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(SCREAMING_SNAKE_CASE    ) - 1) // 2

											return candidates[min_viable_chunk_size_index]


					def 				SCREAMING_SNAKE_CASE_	(    self						:		Optional[int]  ,     SCREAMING_SNAKE_CASE						:		Iterable  ,     SCREAMING_SNAKE_CASE						:		Iterable    )      ->						bool:




											'''simple docstring'''

											__snake_case				=		True
											for aa, aa in zip(SCREAMING_SNAKE_CASE  ,     SCREAMING_SNAKE_CASE    ):
																	assert type(SCREAMING_SNAKE_CASE    ) == type(SCREAMING_SNAKE_CASE    )
																	if isinstance(SCREAMING_SNAKE_CASE  ,     (list, tuple)    ):
																							consistent &= self._compare_arg_caches(SCREAMING_SNAKE_CASE  ,     SCREAMING_SNAKE_CASE    )
																	elif isinstance(SCREAMING_SNAKE_CASE  ,     SCREAMING_SNAKE_CASE    ):
																							__snake_case				=		[v for _, v in sorted(aa.items()  ,     key=lambda SCREAMING_SNAKE_CASE    : x[0]    )]
																							__snake_case				=		[v for _, v in sorted(aa.items()  ,     key=lambda SCREAMING_SNAKE_CASE    : x[0]    )]
																							consistent &= self._compare_arg_caches(SCREAMING_SNAKE_CASE  ,     SCREAMING_SNAKE_CASE    )
																	else:
																							consistent &= aa == aa

											return consistent






					def 				SCREAMING_SNAKE_CASE_	(    self						:		Tuple  ,     SCREAMING_SNAKE_CASE						:		Callable  ,     SCREAMING_SNAKE_CASE						:		tuple  ,     SCREAMING_SNAKE_CASE						:		int  ,     )      ->						int:




											'''simple docstring'''

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

											if not consistent:
																	__snake_case				=		self._determine_favorable_chunk_size(
																	    SCREAMING_SNAKE_CASE  ,     SCREAMING_SNAKE_CASE  ,     SCREAMING_SNAKE_CASE  ,     )
																	__snake_case				=		arg_data

											assert self.cached_chunk_size is not None

											return self.cached_chunk_size









from typing import TYPE_CHECKING

from ....utils import OptionalDependencyNotAvailable, _LazyModule, is_torch_available, is_vision_available


A				:       Optional[Any]      =  {'configuration_van': ['VAN_PRETRAINED_CONFIG_ARCHIVE_MAP', 'VanConfig']}


try:
	if not is_torch_available():
		raise OptionalDependencyNotAvailable()
except OptionalDependencyNotAvailable:
	pass
else:
	A				:       Optional[Any]      =  [
	    'VAN_PRETRAINED_MODEL_ARCHIVE_LIST',
	    'VanForImageClassification',
	    'VanModel',
	    'VanPreTrainedModel',
	]

if TYPE_CHECKING:
	from .configuration_van import VAN_PRETRAINED_CONFIG_ARCHIVE_MAP, VanConfig

	try:
		if not is_torch_available():
			raise OptionalDependencyNotAvailable()
	except OptionalDependencyNotAvailable:
		pass
	else:
		from .modeling_van import (
		    VAN_PRETRAINED_MODEL_ARCHIVE_LIST,
		    VanForImageClassification,
		    VanModel,
		    VanPreTrainedModel,
		)

else:
	import sys

	A				:       Union[str, Any]      =  _LazyModule(__name__, globals()['__file__'], _import_structure)





