


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

	return tuple(reversed(UpperCamelCase__				)				)






@torch.jit.ignore
def 					_A			(			snake_case__    :			Sequence[int]						,						snake_case__    :			Sequence[int]						,						snake_case__    :			Sequence[int]						,						snake_case__    :			Optional[Sequence[bool]] = None						,						snake_case__    :			Optional[Sequence[bool]] = None						,						):







	def reduce_edge_list(snake_case__    :			List[bool]				) -> None:
		snake_case__			:			Union[str, Any]             =		True
		for i in range(len(UpperCamelCase__				)				):
			snake_case__			:			Union[str, Any]             =		-1 * (i + 1)
			l[reversed_idx] &= tally
			snake_case__			:			List[Any]             =		l[reversed_idx]

	if start_edges is None:
		snake_case__			:			Optional[Any]             =		[s == 0 for s in start]
		reduce_edge_list(UpperCamelCase__				)
	if end_edges is None:
		snake_case__			:			Optional[int]             =		[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__			:			List[str]             =		[]
	snake_case__			:			int             =		[]

	# 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__			:			List[Any]             =		tuple(UpperCamelCase__				)
	snake_case__			:			Dict             =		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__			:			Any             =		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__			:			Tuple             =		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__			:			Tuple             =		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			(			snake_case__    :			torch.Tensor						,						snake_case__    :			int						,						snake_case__    :			int						,						snake_case__    :			int				):
	snake_case__			:			int             =		t.shape[:no_batch_dims]
	snake_case__			:			List[Any]             =		list(_flat_idx_to_idx(UpperCamelCase__						,						UpperCamelCase__				)				)
	# _get_minimal_slice_set is inclusive
	snake_case__			:			Tuple             =		list(_flat_idx_to_idx(flat_end - 1						,						UpperCamelCase__				)				)

	# Get an ordered list of slices to perform
	snake_case__			:			Dict             =		_get_minimal_slice_set(
	    UpperCamelCase__						,						UpperCamelCase__						,						UpperCamelCase__						,						)

	snake_case__			:			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 					_A			(			snake_case__    :			Callable						,						snake_case__    :			Dict[str, Any]						,						snake_case__    :			int						,						snake_case__    :			int						,						snake_case__    :			bool = False						,						snake_case__    :			Any = None						,						snake_case__    :			bool = False						,						):







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

	snake_case__			:			Optional[int]             =		[shape[:no_batch_dims] for shape in _fetch_dims(UpperCamelCase__				)]
	snake_case__			:			Tuple             =		tuple([max(UpperCamelCase__				) for s in zip(*UpperCamelCase__				)]				)

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

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

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

	snake_case__			:			List[Any]             =		flat_batch_dim // chunk_size + (flat_batch_dim % chunk_size != 0)

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

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

		snake_case__			:			List[Any]             =		tensor_tree_map(UpperCamelCase__						,						UpperCamelCase__				)

		# Run the layer on the chunk
		snake_case__			:			Tuple             =		layer(**UpperCamelCase__				)

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

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

			def assign(snake_case__    :			dict						,						snake_case__    :			dict				) -> 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__			:			Dict             =		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__			:			Tuple             =		xa
		elif isinstance(UpperCamelCase__						,						torch.Tensor				):
			if _add_into_out:
				out[i : i + chunk_size] += output_chunk
			else:
				snake_case__			:			Optional[int]             =		output_chunk
		else:
			raise ValueError('''Not supported'''				)

		i += chunk_size

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

	return out





class 		snake_case				:

				"""simple docstring"""




				def __init__(	self       ,    lowerCamelCase = 512       ,    )    ->      str:







					"""simple docstring"""






					snake_case__			:			Tuple             =		max_chunk_size
					snake_case__			:			List[str]             =		None
					snake_case__			:			Optional[Any]             =		None





				def    lowercase__							(	self       ,    lowerCamelCase       ,    lowerCamelCase       ,    lowerCamelCase )    ->      List[Any]:







					"""simple docstring"""






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

					if min_chunk_size >= self.max_chunk_size:
						return min_chunk_size

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

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

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

					return candidates[min_viable_chunk_size_index]





				def    lowercase__							(	self       ,    lowerCamelCase       ,    lowerCamelCase )    ->      Optional[int]:







					"""simple docstring"""






					snake_case__			:			Optional[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_ ):
							snake_case__			:			Dict             =		[v for _, v in sorted(aa.items()       ,    key=lambda lowerCamelCase : x[0] )]
							snake_case__			:			List[Any]             =		[v for _, v in sorted(aa.items()       ,    key=lambda lowerCamelCase : x[0] )]
							consistent &= self._compare_arg_caches(A_       ,    A_ )
						else:
							consistent &= aa == aa

					return consistent





				def    lowercase__							(	self       ,    lowerCamelCase       ,    lowerCamelCase       ,    lowerCamelCase       ,    )    ->      Dict:







					"""simple docstring"""






					snake_case__			:			Optional[Any]             =		True
					snake_case__			:			List[Any]             =		tree_map(lambda lowerCamelCase : 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_ )
						snake_case__			:			str             =		self._compare_arg_caches(self.cached_arg_data       ,    A_ )
					else:
						# Otherwise, we can reuse the precomputed value
						snake_case__			:			int             =		False

					if not consistent:
						snake_case__			:			Dict             =		self._determine_favorable_chunk_size(
						    A_       ,    A_       ,    A_       ,    )
						snake_case__			:			List[str]             =		arg_data

					assert self.cached_chunk_size is not None

					return self.cached_chunk_size









"""simple docstring"""

import pytest
import requests

from datasets.utils.file_utils import http_head

from .utils import OfflineSimulationMode, RequestWouldHangIndefinitelyError, offline


@pytest.mark.integration
def lowerCAmelCase_   (					):






 """simple docstring"""







 with offline(OfflineSimulationMode.CONNECTION_TIMES_OUT			):
  with pytest.raises(UpperCamelCase__			):
   requests.request("""GET"""					, """https://huggingface.co"""			)
  with pytest.raises(requests.exceptions.ConnectTimeout			):
   requests.request("""GET"""					, """https://huggingface.co"""					, timeout=1.0			)


@pytest.mark.integration
def lowerCAmelCase_   (					):






 """simple docstring"""







 with offline(OfflineSimulationMode.CONNECTION_FAILS			):
  with pytest.raises(requests.exceptions.ConnectionError			):
   requests.request("""GET"""					, """https://huggingface.co"""			)


def lowerCAmelCase_   (					):






 """simple docstring"""







 with offline(OfflineSimulationMode.HF_DATASETS_OFFLINE_SET_TO_1			):
  with pytest.raises(UpperCamelCase__			):
   http_head("""https://huggingface.co"""			)
