

'''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 	snake_case_					(UpperCamelCase :						Optional[int]   ):





							'''simple docstring'''


							_a			  =       []
							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 	snake_case_					(UpperCamelCase :						Optional[Any]				,		UpperCamelCase :						Tuple   ):





							'''simple docstring'''


							_a			  =       []
							for d in reversed(SCREAMING_SNAKE_CASE_   ):
														idx.append(flat_idx % d   )
														_a			  =       flat_idx // d

							return tuple(reversed(SCREAMING_SNAKE_CASE_   )   )
@torch.jit.ignore
def 	snake_case_					(UpperCamelCase :						Union[str, Any]				,		UpperCamelCase :						Any				,		UpperCamelCase :						str				,		UpperCamelCase :						List[str] = None				,		UpperCamelCase :						str = None				,		):





							'''simple docstring'''


							def reduce_edge_list(UpperCamelCase :						Tuple   ) -> None:
														_a			  =       True
														for i in range(len(SCREAMING_SNAKE_CASE_   )   ):
																					_a			  =       -1 * (i + 1)
																					l[reversed_idx] &= tally
																					_a			  =       l[reversed_idx]

							if start_edges is None:
														_a			  =       [s == 0 for s in start]
														reduce_edge_list(SCREAMING_SNAKE_CASE_   )
							if end_edges is None:
														_a			  =       [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   ),)]

							_a			  =       []
							_a			  =       []

							# 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

							_a			  =       tuple(SCREAMING_SNAKE_CASE_   )
							_a			  =       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

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

														_a			  =       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()   )
														_a			  =       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 	snake_case_					(UpperCamelCase :						Dict				,		UpperCamelCase :						int				,		UpperCamelCase :						Union[str, Any]				,		UpperCamelCase :						Any   ):





							'''simple docstring'''


							_a			  =       t.shape[:no_batch_dims]
							_a			  =       list(_flat_idx_to_idx(SCREAMING_SNAKE_CASE_				,		SCREAMING_SNAKE_CASE_   )   )
							# _get_minimal_slice_set is inclusive
							_a			  =       list(_flat_idx_to_idx(flat_end - 1				,		SCREAMING_SNAKE_CASE_   )   )

							# Get an ordered list of slices to perform
							_a			  =       _get_minimal_slice_set(
							    SCREAMING_SNAKE_CASE_				,		SCREAMING_SNAKE_CASE_				,		SCREAMING_SNAKE_CASE_				,		)

							_a			  =       [t[s] for s in slices]

							return torch.cat([s.view((-1,) + t.shape[no_batch_dims:]   ) for s in sliced_tensors]   )
def 	snake_case_					(UpperCamelCase :						Optional[Any]				,		UpperCamelCase :						Tuple				,		UpperCamelCase :						Optional[int]				,		UpperCamelCase :						Optional[int]				,		UpperCamelCase :						str = False				,		UpperCamelCase :						Dict = None				,		UpperCamelCase :						List[Any] = False				,		):





							'''simple docstring'''


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

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

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

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

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

							_a			  =       flat_batch_dim // chunk_size + (flat_batch_dim % chunk_size != 0)

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

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

														_a			  =       tensor_tree_map(SCREAMING_SNAKE_CASE_				,		SCREAMING_SNAKE_CASE_   )

														# Run the layer on the chunk
														_a			  =       layer(**SCREAMING_SNAKE_CASE_   )

														# Allocate space for the output
														if out is None:
																					_a			  =       tensor_tree_map(lambda UpperCamelCase   : 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(UpperCamelCase :						Union[str, Any]				,		UpperCamelCase :						str   ) -> 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:
																																																	_a			  =       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:
																																			_a			  =       xa
														elif isinstance(SCREAMING_SNAKE_CASE_				,		torch.Tensor   ):
																					if _add_into_out:
																												out[i : i + chunk_size] += output_chunk
																					else:
																												_a			  =       output_chunk
														else:
																					raise ValueError('''Not supported'''   )

														i += chunk_size

							_a			  =       tensor_tree_map(lambda UpperCamelCase   : t.view(orig_batch_dims + t.shape[1:]   )				,		SCREAMING_SNAKE_CASE_   )

							return out

class     A					:







							def __init__(						self :						Optional[Any]							,    lowerCAmelCase_ :						int = 5_12							,    )    ->							List[str]:


														"""simple docstring"""


														_a			  =       max_chunk_size
														_a			  =       None
														_a			  =       None







							def      __lowerCAmelCase    (						self :						Dict							,    lowerCAmelCase_ :						Callable							,    lowerCAmelCase_ :						tuple							,    lowerCAmelCase_ :						int      )    ->							Optional[int]:


														"""simple docstring"""


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

														if min_chunk_size >= self.max_chunk_size:
																					return min_chunk_size

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

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

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

														return candidates[min_viable_chunk_size_index]







							def      __lowerCAmelCase    (						self :						Optional[int]							,    lowerCAmelCase_ :						Iterable							,    lowerCAmelCase_ :						Iterable      )    ->							str:


														"""simple docstring"""


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

														return consistent







							def      __lowerCAmelCase    (						self :						int							,    lowerCAmelCase_ :						Callable							,    lowerCAmelCase_ :						tuple							,    lowerCAmelCase_ :						int							,    )    ->							Tuple:


														"""simple docstring"""


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

														if not consistent:
																					_a			  =       self._determine_favorable_chunk_size(
																					    __lowercase							,    __lowercase							,    __lowercase							,    )
																					_a			  =       arg_data

														assert self.cached_chunk_size is not None

														return self.cached_chunk_size







'''simple docstring'''


from manim import *

class     A					(	_a					):







							def      __lowerCAmelCase    (						self :						Tuple      )    ->							Dict:


														"""simple docstring"""


														_a			  =       Rectangle(height=0.5							,    width=0.5      )
														_a			  =       Rectangle(height=0.4_6							,    width=0.4_6      ).set_stroke(width=0      )

														_a			  =       [mem.copy() for i in range(6      )]
														_a			  =       [mem.copy() for i in range(6      )]
														_a			  =       VGroup(*lowerCAmelCase_      ).arrange(lowerCAmelCase_							,    buff=0      )
														_a			  =       VGroup(*lowerCAmelCase_      ).arrange(lowerCAmelCase_							,    buff=0      )
														_a			  =       VGroup(lowerCAmelCase_							,    lowerCAmelCase_      ).arrange(lowerCAmelCase_							,    buff=0      )
														_a			  =       Text('''CPU'''							,    font_size=24      )
														_a			  =       Group(lowerCAmelCase_							,    lowerCAmelCase_      ).arrange(lowerCAmelCase_							,    buff=0.5							,    aligned_edge=lowerCAmelCase_      )
														cpu.move_to([-2.5, -0.5, 0]      )
														self.add(lowerCAmelCase_      )

														_a			  =       [mem.copy() for i in range(4      )]
														_a			  =       VGroup(*lowerCAmelCase_      ).arrange(lowerCAmelCase_							,    buff=0      )
														_a			  =       Text('''GPU'''							,    font_size=24      )
														_a			  =       Group(lowerCAmelCase_							,    lowerCAmelCase_      ).arrange(lowerCAmelCase_							,    buff=0.5							,    aligned_edge=lowerCAmelCase_      )
														gpu.move_to([-1, -1, 0]      )
														self.add(lowerCAmelCase_      )

														_a			  =       [mem.copy() for i in range(6      )]
														_a			  =       VGroup(*lowerCAmelCase_      ).arrange(lowerCAmelCase_							,    buff=0      )

														_a			  =       Text('''Model'''							,    font_size=24      )
														_a			  =       Group(lowerCAmelCase_							,    lowerCAmelCase_      ).arrange(lowerCAmelCase_							,    buff=0.5							,    aligned_edge=lowerCAmelCase_      )
														model.move_to([3, -1.0, 0]      )
														self.add(lowerCAmelCase_      )

														_a			  =       []
														for i, rect in enumerate(lowerCAmelCase_      ):
																					rect.set_stroke(lowerCAmelCase_      )
																					# target = fill.copy().set_fill(YELLOW, opacity=0.7)
																					# target.move_to(rect)
																					# self.add(target)

																					_a			  =       Rectangle(height=0.4_6 / 4							,    width=0.4_6 / 3      ).set_stroke(width=0.0      ).set_fill(lowerCAmelCase_							,    opacity=0.7      )

																					if i == 0:
																												cpu_target.next_to(cpu_left_col_base[0].get_corner(DOWN + LEFT      )							,    buff=0.0_2							,    direction=lowerCAmelCase_      )
																												cpu_target.set_x(cpu_target.get_x() + 0.1      )
																					elif i == 3:
																												cpu_target.next_to(cpu_targs[0]							,    direction=lowerCAmelCase_							,    buff=0.0      )
																					else:
																												cpu_target.next_to(cpu_targs[i - 1]							,    direction=lowerCAmelCase_							,    buff=0.0      )
																					self.add(lowerCAmelCase_      )
																					cpu_targs.append(lowerCAmelCase_      )

														_a			  =       [mem.copy() for i in range(6      )]
														_a			  =       VGroup(*lowerCAmelCase_      ).arrange(lowerCAmelCase_							,    buff=0      )

														_a			  =       Text('''Loaded Checkpoint'''							,    font_size=24      )
														_a			  =       Group(lowerCAmelCase_							,    lowerCAmelCase_      ).arrange(lowerCAmelCase_							,    aligned_edge=lowerCAmelCase_							,    buff=0.4      )
														checkpoint.move_to([3, 0.5, 0]      )

														_a			  =       Square(side_length=2.2      )
														key.move_to([-5, 2, 0]      )

														_a			  =       MarkupText(
														    F'<b>Key:</b>\n\n<span fgcolor=\'{YELLOW}\'>●</span> Empty Model'							,    font_size=18							,    )

														key_text.move_to([-5, 2.4, 0]      )

														self.add(lowerCAmelCase_							,    lowerCAmelCase_      )

														_a			  =       MarkupText(
														    F'<span fgcolor=\'{BLUE}\'>●</span> Checkpoint'							,    font_size=18							,    )

														blue_text.next_to(lowerCAmelCase_							,    DOWN * 2.4							,    aligned_edge=key_text.get_left()      )

														_a			  =       MarkupText(
														    F'Next, a <i><span fgcolor="{BLUE}">second</span></i> model is loaded into memory,\nwith the weights of a <span fgcolor="{BLUE}">single shard</span>.'							,    font_size=24							,    )
														step_a.move_to([2, 2, 0]      )
														self.play(Write(lowerCAmelCase_      )							,    Write(lowerCAmelCase_      )      )

														self.play(Write(lowerCAmelCase_							,    run_time=1      )							,    Create(lowerCAmelCase_							,    run_time=1      )      )

														_a			  =       []
														_a			  =       []
														for i, rect in enumerate(lowerCAmelCase_      ):
																					_a			  =       fill.copy().set_fill(lowerCAmelCase_							,    opacity=0.7      )
																					target.move_to(lowerCAmelCase_      )
																					first_animations.append(GrowFromCenter(lowerCAmelCase_							,    run_time=1      )      )

																					_a			  =       target.copy()
																					cpu_target.generate_target()
																					if i < 5:
																												cpu_target.target.move_to(cpu_left_col_base[i + 1]      )
																					else:
																												cpu_target.target.move_to(cpu_right_col_base[i - 5]      )
																					second_animations.append(MoveToTarget(lowerCAmelCase_							,    run_time=1.5      )      )

														self.play(*lowerCAmelCase_      )
														self.play(*lowerCAmelCase_      )
														self.wait()

