


'''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 (__lowerCAmelCase   ):
		_UpperCAmelCase		:       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  __lowerCAmelCase (__lowerCAmelCase      , __lowerCAmelCase   ):
		_UpperCAmelCase		:       List[Any]								=      []
		for d in reversed(__lowerCAmelCase   ):
				idx.append(flat_idx % d   )
				_UpperCAmelCase		:       Dict								=      flat_idx // d

		return tuple(reversed(__lowerCAmelCase   )   )






@torch.jit.ignore
def  __lowerCAmelCase (__lowerCAmelCase      , __lowerCAmelCase      , __lowerCAmelCase      , __lowerCAmelCase = None      , __lowerCAmelCase = 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   ) -> None:
				_UpperCAmelCase		:       List[Any]								=      True
				for i in range(len(__lowerCAmelCase   )   ):
						_UpperCAmelCase		:       Tuple								=      -1 * (i + 1)
						l[reversed_idx] &= tally
						_UpperCAmelCase		:       str								=      l[reversed_idx]

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

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

		_UpperCAmelCase		:       Tuple[slice, ...]								=      tuple(__lowerCAmelCase   )
		_UpperCAmelCase		:       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

				_UpperCAmelCase		:       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

				_UpperCAmelCase		:       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()   )
				_UpperCAmelCase		:       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  __lowerCAmelCase (__lowerCAmelCase      , __lowerCAmelCase      , __lowerCAmelCase      , __lowerCAmelCase   ):
		_UpperCAmelCase		:       Any								=      t.shape[:no_batch_dims]
		_UpperCAmelCase		:       Dict								=      list(_flat_idx_to_idx(__lowerCAmelCase      , __lowerCAmelCase   )   )
		# _get_minimal_slice_set is inclusive
		_UpperCAmelCase		:       Optional[int]								=      list(_flat_idx_to_idx(flat_end - 1      , __lowerCAmelCase   )   )

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

		_UpperCAmelCase		:       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  __lowerCAmelCase (__lowerCAmelCase      , __lowerCAmelCase      , __lowerCAmelCase      , __lowerCAmelCase      , __lowerCAmelCase = False      , __lowerCAmelCase = None      , __lowerCAmelCase = False      , ):
		if not (len(__lowerCAmelCase   ) > 0):
				raise ValueError("Must provide at least one input"   )

		_UpperCAmelCase		:       int								=      [shape[:no_batch_dims] for shape in _fetch_dims(__lowerCAmelCase   )]
		_UpperCAmelCase		:       Optional[int]								=      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:
								_UpperCAmelCase		:       Optional[Any]								=      t.expand(orig_batch_dims + t.shape[no_batch_dims:]   )
						_UpperCAmelCase		:       str								=      t.reshape(-1      , *t.shape[no_batch_dims:]   )
				else:
						_UpperCAmelCase		:       List[str]								=      t.expand(orig_batch_dims + t.shape[no_batch_dims:]   )
				return t

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

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

		_UpperCAmelCase		:       List[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

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

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

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

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

				i += chunk_size

		_UpperCAmelCase		:       Union[str, Any]								=      tensor_tree_map(lambda __lowerCAmelCase   : t.view(orig_batch_dims + t.shape[1:]   )      , __lowerCAmelCase   )

		return out






class 	lowerCAmelCase__			:




			def __init__(		self  :     Tuple	,  lowerCamelCase__  :     int = 5_12	,  )   ->str:



					'''simple docstring'''







					_UpperCAmelCase		:       str								=      max_chunk_size
					_UpperCAmelCase		:       Optional[int]								=      None
					_UpperCAmelCase		:       Optional[tuple]								=      None




			def 							lowerCAmelCase__    (		self  :     Union[str, Any]	,  lowerCamelCase__  :     Callable	,  lowerCamelCase__  :     tuple	,  lowerCamelCase__  :     int )   ->int:



					'''simple docstring'''







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

					if min_chunk_size >= self.max_chunk_size:
							return min_chunk_size

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

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

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

					return candidates[min_viable_chunk_size_index]




			def 							lowerCAmelCase__    (		self  :     Union[str, Any]	,  lowerCamelCase__  :     Iterable	,  lowerCamelCase__  :     Iterable )   ->bool:



					'''simple docstring'''







					_UpperCAmelCase		:       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__ ):
									_UpperCAmelCase		:       str								=      [v for _, v in sorted(aa.items()	,  key=lambda lowerCamelCase__ : x[0] )]
									_UpperCAmelCase		:       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 							lowerCAmelCase__    (		self  :     List[str]	,  lowerCamelCase__  :     Callable	,  lowerCamelCase__  :     tuple	,  lowerCamelCase__  :     int	,  )   ->int:



					'''simple docstring'''







					_UpperCAmelCase		:       Optional[Any]								=      True
					_UpperCAmelCase		:       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__ )
							_UpperCAmelCase		:       Dict								=      self._compare_arg_caches(self.cached_arg_data	,  lowerCamelCase__ )
					else:
							# Otherwise, we can reuse the precomputed value
							_UpperCAmelCase		:       int								=      False

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

					assert self.cached_chunk_size is not None

					return self.cached_chunk_size







'''simple docstring'''


from typing import Optional
from urllib.parse import quote

import huggingface_hub as hfh
from packaging import version






def  __lowerCAmelCase (__lowerCAmelCase      , __lowerCAmelCase      , __lowerCAmelCase = None   ):
		if version.parse(hfh.__version__   ).release < version.parse("0.11.0"   ).release:
				# old versions of hfh don't url-encode the file path
				_UpperCAmelCase		:       str								=      quote(__lowerCAmelCase   )
		return hfh.hf_hub_url(__lowerCAmelCase      , __lowerCAmelCase      , repo_type="dataset"      , revision=__lowerCAmelCase   )



