




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 	lowercase_		(     SCREAMING_SNAKE_CASE						:		Union[dict, list, tuple, torch.Tensor] ):






	"""simple docstring"""
	snake_case__						:     Tuple		=[]
	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 	lowercase_		(     SCREAMING_SNAKE_CASE						:		int						,						SCREAMING_SNAKE_CASE						:		Tuple[int, ...] ):






	"""simple docstring"""
	snake_case__						:     Optional[int]		=[]
	for d in reversed(UpperCamelCase__ ):
		idx.append(flat_idx % d )
		snake_case__						:     Any		=flat_idx // d

	return tuple(reversed(UpperCamelCase__ ) )







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






	"""simple docstring"""

	# 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(SCREAMING_SNAKE_CASE						:		List[bool] ) -> None:
		snake_case__						:     List[str]		=True
		for i in range(len(UpperCamelCase__ ) ):
			snake_case__						:     Union[str, Any]		=-1 * (i + 1)
			l[reversed_idx] &= tally
			snake_case__						:     Any		=l[reversed_idx]

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

	# 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__						:     Optional[Any]		=tuple(UpperCamelCase__ )
	snake_case__						:     Any		=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__						:     Dict		=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__						:     Any		=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__						:     Optional[int]		=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 	lowercase_		(     SCREAMING_SNAKE_CASE						:		torch.Tensor						,						SCREAMING_SNAKE_CASE						:		int						,						SCREAMING_SNAKE_CASE						:		int						,						SCREAMING_SNAKE_CASE						:		int ):






	"""simple docstring"""
	snake_case__						:     Union[str, Any]		=t.shape[:no_batch_dims]
	snake_case__						:     Optional[int]		=list(_flat_idx_to_idx(UpperCamelCase__						,						UpperCamelCase__ ) )
	# _get_minimal_slice_set is inclusive
	snake_case__						:     str		=list(_flat_idx_to_idx(flat_end - 1						,						UpperCamelCase__ ) )

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

	snake_case__						:     Optional[int]		=[t[s] for s in slices]

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







def 	lowercase_		(     SCREAMING_SNAKE_CASE						:		Callable						,						SCREAMING_SNAKE_CASE						:		Dict[str, Any]						,						SCREAMING_SNAKE_CASE						:		int						,						SCREAMING_SNAKE_CASE						:		int						,						SCREAMING_SNAKE_CASE						:		bool = False						,						SCREAMING_SNAKE_CASE						:		Any = None						,						SCREAMING_SNAKE_CASE						:		bool = False						,						):






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

	snake_case__						:     List[str]		=[shape[:no_batch_dims] for shape in _fetch_dims(UpperCamelCase__ )]
	snake_case__						:     Optional[Any]		=tuple([max(UpperCamelCase__ ) for s in zip(*UpperCamelCase__ )] )

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

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

	snake_case__						:     List[str]		=1
	for d in orig_batch_dims:
		flat_batch_dim *= d

	snake_case__						:     Tuple		=flat_batch_dim // chunk_size + (flat_batch_dim % chunk_size != 0)

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

	snake_case__						:     Optional[Any]		=0
	snake_case__						:     Dict		=prepped_outputs
	for _ in range(UpperCamelCase__ ):
		# Chunk the input
		if not low_mem:
			snake_case__						:     Dict		=_select_chunk
		else:
			snake_case__						:     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__						:     Optional[int]		=layer(**UpperCamelCase__ )

		# Allocate space for the output
		if out is None:
			snake_case__						:     Any		=tensor_tree_map(lambda SCREAMING_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(SCREAMING_SNAKE_CASE						:		dict						,						SCREAMING_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__						:     Any		=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__						:     List[Any]		=xa
		elif isinstance(UpperCamelCase__						,						torch.Tensor ):
			if _add_into_out:
				out[i : i + chunk_size] += output_chunk
			else:
				snake_case__						:     int		=output_chunk
		else:
			raise ValueError('''Not supported''' )

		i += chunk_size

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

	return out






class 		_lowerCAmelCase       :
			"""simple docstring"""
			def __init__(					self     ,		__SCREAMING_SNAKE_CASE = 512     ,		)		->							Tuple:





				"""simple docstring"""







				snake_case__						:     int		=max_chunk_size
				snake_case__						:     str		=None
				snake_case__						:     List[str]		=None
			def 					UpperCAmelCase     (					self     ,		__SCREAMING_SNAKE_CASE     ,		__SCREAMING_SNAKE_CASE     ,		__SCREAMING_SNAKE_CASE       )		->							int:





				"""simple docstring"""







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

				if min_chunk_size >= self.max_chunk_size:
					return min_chunk_size

				snake_case__						:     List[Any]		=[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__						:     Tuple		=[min_chunk_size] + candidates
				candidates[-1] += 4

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

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

				return candidates[min_viable_chunk_size_index]
			def 					UpperCAmelCase     (					self     ,		__SCREAMING_SNAKE_CASE     ,		__SCREAMING_SNAKE_CASE       )		->							bool:





				"""simple docstring"""







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

				return consistent


			def 					UpperCAmelCase     (					self     ,		__SCREAMING_SNAKE_CASE     ,		__SCREAMING_SNAKE_CASE     ,		__SCREAMING_SNAKE_CASE     ,		)		->							int:





				"""simple docstring"""







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

				if not consistent:
					snake_case__						:     Dict		=self._determine_favorable_chunk_size(
					    __A     ,		__A     ,		__A     ,		)
					snake_case__						:     str		=arg_data

				assert self.cached_chunk_size is not None

				return self.cached_chunk_size











from ...processing_utils import ProcessorMixin
from ...tokenization_utils_base import BatchEncoding




class 	UpperCamelCase_							(  UpperCamelCase__     ):
							lowerCamelCase_							     =       ["image_processor", "tokenizer"]
							lowerCamelCase_							     =       "AutoImageProcessor"
							lowerCamelCase_							     =       "AutoTokenizer"


							def __init__(				self     :Optional[int]				,					__A     :Optional[Any]				,					__A     :Dict		) -> Dict:



									"""simple docstring"""



									super().__init__(__A				,					__A		)
									SCREAMING_SNAKE_CASE__								=   self.image_processor


							def __call__(				self     :int				,					__A     :str=None				,					__A     :int=None				,					__A     :Union[str, Any]=None				,					**__A     :str		) -> Optional[Any]:



									"""simple docstring"""



									if text is None and images is None:
											raise ValueError("""You have to specify either text or images. Both cannot be none."""		)

									if text is not None:
											SCREAMING_SNAKE_CASE__								=   self.tokenizer(__A				,					return_tensors=__A				,					**__A		)

									if images is not None:
											SCREAMING_SNAKE_CASE__								=   self.image_processor(__A				,					return_tensors=__A				,					**__A		)

									if text is not None and images is not None:
											SCREAMING_SNAKE_CASE__								=   image_features.pixel_values
											return encoding
									elif text is not None:
											return encoding
									else:
											return BatchEncoding(data=dict(**__A		)				,					tensor_type=__A		)


							def   _snake_case	(				self     :str				,					*__A     :List[str]				,					**__A     :List[str]		) -> List[Any]:



									"""simple docstring"""



									return self.tokenizer.batch_decode(*__A				,					**__A		)


							def   _snake_case	(				self     :List[str]				,					*__A     :Any				,					**__A     :Any		) -> Tuple:



									"""simple docstring"""



									return self.tokenizer.decode(*__A				,					**__A		)


							@property
							def   _snake_case	(				self     :Dict		) -> List[Any]:



									"""simple docstring"""



									return ["input_ids", "attention_mask", "pixel_values"]