
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]					):


				'''simple docstring'''
				lowercase_       =					[]
				if isinstance(snake_case__						,   snake_case__					):
								for v in tree.values():
												shapes.extend(_fetch_dims(snake_case__					)					)
				elif isinstance(snake_case__						,   (list, tuple)					):
								for t in tree:
												shapes.extend(_fetch_dims(snake_case__					)					)
				elif isinstance(snake_case__						,   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, ...]					):


				'''simple docstring'''
				lowercase_       =					[]
				for d in reversed(snake_case__					):
								idx.append(flat_idx % d					)
								lowercase_       =					flat_idx // d

				return tuple(reversed(snake_case__					)					)




@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						,   ):


				'''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(snake_case__:					List[bool]					) -> None:
								lowercase_       =					True
								for i in range(len(snake_case__					)					):
												lowercase_       =					-1 * (i + 1)
												l[reversed_idx] &= tally
												lowercase_       =					l[reversed_idx]

				if start_edges is None:
								lowercase_       =					[s == 0 for s in start]
								reduce_edge_list(snake_case__					)
				if end_edges is None:
								lowercase_       =					[e == (d - 1) for e, d in zip(snake_case__						,   snake_case__					)]
								reduce_edge_list(snake_case__					)

				# Base cases. Either start/end are empty and we're done, or the final,
				# one-dimensional tensor can be simply sliced
				if len(snake_case__					) == 0:
								return [()]
				elif len(snake_case__					) == 1:
								return [(slice(start[0]						,   end[0] + 1					),)]

				lowercase_       =					[]
				lowercase_       =					[]

				# Dimensions common to start and end can be selected directly
				for s, e in zip(snake_case__						,   snake_case__					):
								if s == e:
												path_list.append(slice(snake_case__						,   s + 1					)					)
								else:
												break

				lowercase_       =					tuple(snake_case__					)
				lowercase_       =					len(snake_case__					)

				# start == end, and we're done
				if divergence_idx == len(snake_case__					):
								return [path]

				def upper() -> Tuple[Tuple[slice, ...], ...]:
								assert start_edges is not None
								assert end_edges is not None

								lowercase_       =					start[divergence_idx]
								return tuple(
								    path + (slice(snake_case__						,   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_       =					end[divergence_idx]
								return tuple(
								    path + (slice(snake_case__						,   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_       =					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					):


				'''simple docstring'''
				lowercase_       =					t.shape[:no_batch_dims]
				lowercase_       =					list(_flat_idx_to_idx(snake_case__						,   snake_case__					)					)
				# _get_minimal_slice_set is inclusive
				lowercase_       =					list(_flat_idx_to_idx(flat_end - 1						,   snake_case__					)					)

				# Get an ordered list of slices to perform
				lowercase_       =					_get_minimal_slice_set(
				    snake_case__						,   snake_case__						,   snake_case__						,   )

				lowercase_       =					[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						,   ):


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

				lowercase_       =					[shape[:no_batch_dims] for shape in _fetch_dims(snake_case__					)]
				lowercase_       =					tuple([max(snake_case__					) for s in zip(*snake_case__					)]					)

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

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

				lowercase_       =					1
				for d in orig_batch_dims:
								flat_batch_dim *= d

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

				lowercase_       =					0
				lowercase_       =					prepped_outputs
				for _ in range(snake_case__					):
								# Chunk the input
								if not low_mem:
												lowercase_       =					_select_chunk
								else:
												lowercase_       =					partial(
												    _chunk_slice						,   flat_start=snake_case__						,   flat_end=min(snake_case__						,   i + chunk_size					)						,   no_batch_dims=len(snake_case__					)						,   )

								lowercase_       =					tensor_tree_map(snake_case__						,   snake_case__					)

								# Run the layer on the chunk
								lowercase_       =					layer(**snake_case__					)

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

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

												def assign(snake_case__:					dict						,   snake_case__:					dict					) -> None:
																for k, v in da.items():
																				if isinstance(snake_case__						,   snake_case__					):
																								assign(snake_case__						,   da[k]					)
																				else:
																								if _add_into_out:
																												v[i : i + chunk_size] += da[k]
																								else:
																												lowercase_       =					da[k]

												assign(snake_case__						,   snake_case__					)
								elif isinstance(snake_case__						,   snake_case__					):
												for xa, xa in zip(snake_case__						,   snake_case__					):
																if _add_into_out:
																				xa[i : i + chunk_size] += xa
																else:
																				lowercase_       =					xa
								elif isinstance(snake_case__						,   torch.Tensor					):
												if _add_into_out:
																out[i : i + chunk_size] += output_chunk
												else:
																lowercase_       =					output_chunk
								else:
												raise ValueError('''Not supported'''					)

								i += chunk_size

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

				return out


class  lowercase__:



	"""simple docstring"""




	def __init__(       self  :			Union[str, Any]       ,							SCREAMING_SNAKE_CASE_  :			int = 5_1_2       ,							)       ->     Tuple:
					lowercase_       =					max_chunk_size
					lowercase_       =					None
					lowercase_       =					None
	def 						_lowercase  (       self  :			Dict       ,							SCREAMING_SNAKE_CASE_  :			Callable       ,							SCREAMING_SNAKE_CASE_  :			tuple       ,							SCREAMING_SNAKE_CASE_  :			int       )       ->     int:
					logging.info('''Tuning chunk size...'''       )

					if min_chunk_size >= self.max_chunk_size:
									return min_chunk_size

					lowercase_       =					[2**l for l in range(int(math.log(self.max_chunk_size       ,							2       )       ) + 1       )]
					lowercase_       =					[c for c in candidates if c > min_chunk_size]
					lowercase_       =					[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

					lowercase_       =					0
					lowercase_       =					len(SCREAMING_SNAKE_CASE_       ) - 1
					while i > min_viable_chunk_size_index:
									lowercase_       =					test_chunk_size(candidates[i]       )
									if not viable:
													lowercase_       =					(min_viable_chunk_size_index + i) // 2
									else:
													lowercase_       =					i
													lowercase_       =					(i + len(SCREAMING_SNAKE_CASE_       ) - 1) // 2

					return candidates[min_viable_chunk_size_index]
	def 						_lowercase  (       self  :			Tuple       ,							SCREAMING_SNAKE_CASE_  :			Iterable       ,							SCREAMING_SNAKE_CASE_  :			Iterable       )       ->     bool:
					lowercase_       =					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_       ):
													lowercase_       =					[v for _, v in sorted(aa.items()       ,							key=lambda SCREAMING_SNAKE_CASE_       : x[0]       )]
													lowercase_       =					[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 						_lowercase  (       self  :			Tuple       ,							SCREAMING_SNAKE_CASE_  :			Callable       ,							SCREAMING_SNAKE_CASE_  :			tuple       ,							SCREAMING_SNAKE_CASE_  :			int       ,							)       ->     int:
					lowercase_       =					True
					lowercase_       =					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_       )
									lowercase_       =					self._compare_arg_caches(self.cached_arg_data       ,							SCREAMING_SNAKE_CASE_       )
					else:
									# Otherwise, we can reuse the precomputed value
									lowercase_       =					False

					if not consistent:
									lowercase_       =					self._determine_favorable_chunk_size(
									    SCREAMING_SNAKE_CASE_       ,							SCREAMING_SNAKE_CASE_       ,							SCREAMING_SNAKE_CASE_       ,							)
									lowercase_       =					arg_data

					assert self.cached_chunk_size is not None

					return self.cached_chunk_size






import unittest

import numpy as np
import torch

from .utils_summarization import build_mask, compute_token_type_ids, process_story, truncate_or_pad
class  lowercase__(				unittest.TestCase							):



	"""simple docstring"""




	def 						_lowercase  (       self  :			List[str]       )       ->     List[Any]:
					lowercase_       =					1_0
	def 						_lowercase  (       self  :			int       )       ->     List[str]:
					lowercase_       =					[1, 2, 3, 4]
					lowercase_       =					[1, 2, 3, 4, 0, 0, 0, 0, 0, 0]
					self.assertEqual(truncate_or_pad(SCREAMING_SNAKE_CASE_       ,							self.block_size       ,							0       )       ,							SCREAMING_SNAKE_CASE_       )
	def 						_lowercase  (       self  :			int       )       ->     Optional[Any]:
					lowercase_       =					[1, 2, 3, 4, 5, 6, 7, 8, 9, 1_0]
					lowercase_       =					[1, 2, 3, 4, 5, 6, 7, 8, 9, 1_0]
					self.assertEqual(truncate_or_pad(SCREAMING_SNAKE_CASE_       ,							self.block_size       ,							0       )       ,							SCREAMING_SNAKE_CASE_       )
	def 						_lowercase  (       self  :			Union[str, Any]       )       ->     Optional[int]:
					lowercase_       =					[1, 2, 3, 4, 5, 6, 7, 8, 9, 1_0, 1_1, 1_2, 1_3]
					lowercase_       =					[1, 2, 3, 4, 5, 6, 7, 8, 9, 1_0]
					self.assertEqual(truncate_or_pad(SCREAMING_SNAKE_CASE_       ,							self.block_size       ,							0       )       ,							SCREAMING_SNAKE_CASE_       )
	def 						_lowercase  (       self  :			Any       )       ->     List[Any]:
					lowercase_       =					'''It was the year of Our Lord one thousand seven hundred and
        seventy-five.\n\nSpiritual revelations were conceded to England at that
        favoured period, as at this.'''
					lowercase_  ,	lowercase_       =					process_story(SCREAMING_SNAKE_CASE_       )
					self.assertEqual(SCREAMING_SNAKE_CASE_       ,							[]       )
	def 						_lowercase  (       self  :			List[str]       )       ->     List[str]:
					lowercase_       =					''''''
					lowercase_  ,	lowercase_       =					process_story(SCREAMING_SNAKE_CASE_       )
					self.assertEqual(SCREAMING_SNAKE_CASE_       ,							[]       )
					self.assertEqual(SCREAMING_SNAKE_CASE_       ,							[]       )
	def 						_lowercase  (       self  :			Union[str, Any]       )       ->     Union[str, Any]:
					lowercase_       =					(
					    '''It was the year of Our Lord one thousand seven hundred and '''
					    '''seventy-five\n\nSpiritual revelations were conceded to England '''
					    '''at that favoured period, as at this.\n@highlight\n\nIt was the best of times'''
					)
					lowercase_  ,	lowercase_       =					process_story(SCREAMING_SNAKE_CASE_       )

					lowercase_       =					[
					    '''It was the year of Our Lord one thousand seven hundred and seventy-five.''',
					    '''Spiritual revelations were conceded to England at that favoured period, as at this.''',
					]
					self.assertEqual(SCREAMING_SNAKE_CASE_       ,							SCREAMING_SNAKE_CASE_       )

					lowercase_       =					['''It was the best of times.''']
					self.assertEqual(SCREAMING_SNAKE_CASE_       ,							SCREAMING_SNAKE_CASE_       )
	def 						_lowercase  (       self  :			Union[str, Any]       )       ->     Optional[Any]:
					lowercase_       =					torch.tensor([1, 2, 3, 4]       )
					lowercase_       =					torch.tensor([1, 1, 1, 1]       )
					np.testing.assert_array_equal(build_mask(SCREAMING_SNAKE_CASE_       ,							0       ).numpy()       ,							expected.numpy()       )
	def 						_lowercase  (       self  :			List[Any]       )       ->     Tuple:
					lowercase_       =					torch.tensor([1, 2, 3, 4, 2_3, 2_3, 2_3]       )
					lowercase_       =					torch.tensor([1, 1, 1, 1, 0, 0, 0]       )
					np.testing.assert_array_equal(build_mask(SCREAMING_SNAKE_CASE_       ,							2_3       ).numpy()       ,							expected.numpy()       )
	def 						_lowercase  (       self  :			int       )       ->     Dict:
					lowercase_       =					torch.tensor([8, 2, 3, 4, 1, 1, 1]       )
					lowercase_       =					torch.tensor([1, 1, 1, 1, 0, 0, 0]       )
					np.testing.assert_array_equal(build_mask(SCREAMING_SNAKE_CASE_       ,							1       ).numpy()       ,							expected.numpy()       )




	def 						_lowercase  (       self  :			List[str]       )       ->     Tuple:
					lowercase_       =					1_0_1
					lowercase_       =					torch.tensor([[1, 2, 3, 4, 5, 6], [1, 2, 3, 1_0_1, 5, 6], [1, 1_0_1, 3, 4, 1_0_1, 6]]       )
					lowercase_       =					torch.tensor([[1, 1, 1, 1, 1, 1], [1, 1, 1, 0, 0, 0], [1, 0, 0, 0, 1, 1]]       )

					lowercase_       =					compute_token_type_ids(SCREAMING_SNAKE_CASE_       ,							SCREAMING_SNAKE_CASE_       )
					np.testing.assert_array_equal(SCREAMING_SNAKE_CASE_       ,							SCREAMING_SNAKE_CASE_       )




