


"""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				(      _SCREAMING_SNAKE_CASE				)    ->				List[Tuple[int, ...]]:
							snake_case_           =		[]
							if isinstance(_SCREAMING_SNAKE_CASE				, _SCREAMING_SNAKE_CASE				):
														for v in tree.values():
																					shapes.extend(_fetch_dims(_SCREAMING_SNAKE_CASE				)				)
							elif isinstance(_SCREAMING_SNAKE_CASE				, (list, tuple)				):
														for t in tree:
																					shapes.extend(_fetch_dims(_SCREAMING_SNAKE_CASE				)				)
							elif isinstance(_SCREAMING_SNAKE_CASE				, torch.Tensor				):
														shapes.append(tree.shape				)
							else:
														raise ValueError("""Not supported"""				)

							return shapes







@torch.jit.ignore
def 						_a				(      _SCREAMING_SNAKE_CASE				, _SCREAMING_SNAKE_CASE				)    ->				Tuple[int, ...]:
							snake_case_           =		[]
							for d in reversed(_SCREAMING_SNAKE_CASE				):
														idx.append(flat_idx % d				)
														snake_case_           =		flat_idx // d

							return tuple(reversed(_SCREAMING_SNAKE_CASE				)				)







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

							# 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				) -> None:
														snake_case_           =		True
														for i in range(len(_SCREAMING_SNAKE_CASE				)				):
																					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(_SCREAMING_SNAKE_CASE				)
							if end_edges is None:
														snake_case_           =		[e == (d - 1) for e, d in zip(_SCREAMING_SNAKE_CASE				, _SCREAMING_SNAKE_CASE				)]
														reduce_edge_list(_SCREAMING_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(_SCREAMING_SNAKE_CASE				) == 0:
														return [()]
							elif len(_SCREAMING_SNAKE_CASE				) == 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(_SCREAMING_SNAKE_CASE				, _SCREAMING_SNAKE_CASE				):
														if s == e:
																					path_list.append(slice(_SCREAMING_SNAKE_CASE				, s + 1				)				)
														else:
																					break

							snake_case_           =		tuple(_SCREAMING_SNAKE_CASE				)
							snake_case_           =		len(_SCREAMING_SNAKE_CASE				)

							# start == end, and we're done
							if divergence_idx == len(_SCREAMING_SNAKE_CASE				):
														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(_SCREAMING_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

														snake_case_           =		end[divergence_idx]
														return tuple(
														    path + (slice(_SCREAMING_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()				)
														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 						_a				(      _SCREAMING_SNAKE_CASE				, _SCREAMING_SNAKE_CASE				, _SCREAMING_SNAKE_CASE				, _SCREAMING_SNAKE_CASE				)    ->				torch.Tensor:
							snake_case_           =		t.shape[:no_batch_dims]
							snake_case_           =		list(_flat_idx_to_idx(_SCREAMING_SNAKE_CASE				, _SCREAMING_SNAKE_CASE				)				)
							# _get_minimal_slice_set is inclusive
							snake_case_           =		list(_flat_idx_to_idx(flat_end - 1				, _SCREAMING_SNAKE_CASE				)				)

							# Get an ordered list of slices to perform
							snake_case_           =		_get_minimal_slice_set(
							    _SCREAMING_SNAKE_CASE				, _SCREAMING_SNAKE_CASE				, _SCREAMING_SNAKE_CASE				, )

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

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

							def _prep_inputs(_SCREAMING_SNAKE_CASE				) -> 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				, _SCREAMING_SNAKE_CASE				)
							snake_case_           =		None
							if _out is not None:
														snake_case_           =		tensor_tree_map(lambda _SCREAMING_SNAKE_CASE				: 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(_SCREAMING_SNAKE_CASE				) -> 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(_SCREAMING_SNAKE_CASE				):
														# Chunk the input
														if not low_mem:
																					snake_case_           =		_select_chunk
														else:
																					snake_case_           =		partial(
																					    _chunk_slice				, flat_start=_SCREAMING_SNAKE_CASE				, flat_end=min(_SCREAMING_SNAKE_CASE				, i + chunk_size				)				, no_batch_dims=len(_SCREAMING_SNAKE_CASE				)				, )

														snake_case_           =		tensor_tree_map(_SCREAMING_SNAKE_CASE				, _SCREAMING_SNAKE_CASE				)

														# Run the layer on the chunk
														snake_case_           =		layer(**_SCREAMING_SNAKE_CASE				)

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

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

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

																					assign(_SCREAMING_SNAKE_CASE				, _SCREAMING_SNAKE_CASE				)
														elif isinstance(_SCREAMING_SNAKE_CASE				, _SCREAMING_SNAKE_CASE				):
																					for xa, xa in zip(_SCREAMING_SNAKE_CASE				, _SCREAMING_SNAKE_CASE				):
																												if _add_into_out:
																																			xa[i : i + chunk_size] += xa
																												else:
																																			snake_case_           =		xa
														elif isinstance(_SCREAMING_SNAKE_CASE				, 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 _SCREAMING_SNAKE_CASE				: t.view(orig_batch_dims + t.shape[1:]				)				, _SCREAMING_SNAKE_CASE				)

							return out





class      __A     :





							'''simple docstring'''

							def __init__(				self							:      Tuple		,       UpperCAmelCase_							:      int = 512		,       )					->Any:



														"""simple docstring"""



														snake_case_           =		max_chunk_size
														snake_case_           =		None
														snake_case_           =		None



							def 		lowerCAmelCase       (				self							:      List[Any]		,       UpperCAmelCase_							:      Callable		,       UpperCAmelCase_							:      tuple		,       UpperCAmelCase_							:      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(UpperCAmelCase_							:      int       ) -> bool:
																					try:
																												with torch.no_grad():
																																			fn(*UpperCAmelCase_		,       chunk_size=UpperCAmelCase_       )
																												return True
																					except RuntimeError:
																												return False

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

														return candidates[min_viable_chunk_size_index]



							def 		lowerCAmelCase       (				self							:      Dict		,       UpperCAmelCase_							:      Iterable		,       UpperCAmelCase_							:      Iterable       )					->bool:



														"""simple docstring"""



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

														return consistent



							def 		lowerCAmelCase       (				self							:      List[str]		,       UpperCAmelCase_							:      Callable		,       UpperCAmelCase_							:      tuple		,       UpperCAmelCase_							:      int		,       )					->int:



														"""simple docstring"""



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

														if not consistent:
																					snake_case_           =		self._determine_favorable_chunk_size(
																					    UpperCAmelCase_		,       UpperCAmelCase_		,       UpperCAmelCase_		,       )
																					snake_case_           =		arg_data

														assert self.cached_chunk_size is not None

														return self.cached_chunk_size







"""simple docstring"""
from __future__ import annotations

from collections import deque





class      __A     :





							'''simple docstring'''

							def __init__(				self							:      List[Any]		,       UpperCAmelCase_							:      list[str]       )					->List[Any]:



														"""simple docstring"""



														snake_case_           =		[]
														self.adlist.append(
														    {"""value""": """""", """next_states""": [], """fail_state""": 0, """output""": []}       )

														for keyword in keywords:
																					self.add_keyword(UpperCAmelCase_       )
														self.set_fail_transitions()



							def 		lowerCAmelCase       (				self							:      int		,       UpperCAmelCase_							:      int		,       UpperCAmelCase_							:      str       )					->int | None:



														"""simple docstring"""



														for state in self.adlist[current_state]["next_states"]:
																					if char == self.adlist[state]["value"]:
																												return state
														return None



							def 		lowerCAmelCase       (				self							:      int		,       UpperCAmelCase_							:      str       )					->None:



														"""simple docstring"""



														snake_case_           =		0
														for character in keyword:
																					snake_case_           =		self.find_next_state(UpperCAmelCase_		,       UpperCAmelCase_       )
																					if next_state is None:
																												self.adlist.append(
																												    {
																												        """value""": character,
																												        """next_states""": [],
																												        """fail_state""": 0,
																												        """output""": [],
																												    }       )
																												self.adlist[current_state]["next_states"].append(len(self.adlist       ) - 1       )
																												snake_case_           =		len(self.adlist       ) - 1
																					else:
																												snake_case_           =		next_state
														self.adlist[current_state]["output"].append(UpperCAmelCase_       )



							def 		lowerCAmelCase       (				self							:      Optional[Any]       )					->None:



														"""simple docstring"""



														snake_case_           =		deque()
														for node in self.adlist[0]["next_states"]:
																					q.append(UpperCAmelCase_       )
																					snake_case_           =		0
														while q:
																					snake_case_           =		q.popleft()
																					for child in self.adlist[r]["next_states"]:
																												q.append(UpperCAmelCase_       )
																												snake_case_           =		self.adlist[r]["""fail_state"""]
																												while (
																												    self.find_next_state(UpperCAmelCase_		,       self.adlist[child]["""value"""]       ) is None
																												    and state != 0
																												):
																																			snake_case_           =		self.adlist[state]["""fail_state"""]
																												snake_case_           =		self.find_next_state(
																												    UpperCAmelCase_		,       self.adlist[child]["""value"""]       )
																												if self.adlist[child]["fail_state"] is None:
																																			snake_case_           =		0
																												snake_case_           =		(
																												    self.adlist[child]["""output"""]
																												    + self.adlist[self.adlist[child]["""fail_state"""]]["""output"""]
																												)



							def 		lowerCAmelCase       (				self							:      Union[str, Any]		,       UpperCAmelCase_							:      str       )					->dict[str, list[int]]:



														"""simple docstring"""



														snake_case_           =		{}  # returns a dict with keywords and list of its occurrences
														snake_case_           =		0
														for i in range(len(UpperCAmelCase_       )       ):
																					while (
																					    self.find_next_state(UpperCAmelCase_		,       string[i]       ) is None
																					    and current_state != 0
																					):
																												snake_case_           =		self.adlist[current_state]["""fail_state"""]
																					snake_case_           =		self.find_next_state(UpperCAmelCase_		,       string[i]       )
																					if next_state is None:
																												snake_case_           =		0
																					else:
																												snake_case_           =		next_state
																												for key in self.adlist[current_state]["output"]:
																																			if key not in result:
																																										snake_case_           =		[]
																																			result[key].append(i - len(UpperCAmelCase_       ) + 1       )
														return result


if __name__ == "__main__":
				import doctest

				doctest.testmod()



