


'''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		(lowercase__			:   Union[dict, list, tuple, torch.Tensor]						)			->			List[Tuple[int, ...]]:






     """simple docstring"""
     __snake_case           =						[]
     if isinstance(lowercase__							,   lowercase__						):
          for v in tree.values():
               shapes.extend(_fetch_dims(lowercase__						)						)
     elif isinstance(lowercase__							,   (list, tuple)						):
          for t in tree:
               shapes.extend(_fetch_dims(lowercase__						)						)
     elif isinstance(lowercase__							,   torch.Tensor						):
          shapes.append(tree.shape						)
     else:
          raise ValueError('Not supported'						)

     return shapes





@torch.jit.ignore
def   _a		(lowercase__			:   int							,   lowercase__			:   Tuple[int, ...]						)			->			Tuple[int, ...]:






     """simple docstring"""
     __snake_case           =						[]
     for d in reversed(lowercase__						):
          idx.append(flat_idx % d						)
          __snake_case           =						flat_idx // d

     return tuple(reversed(lowercase__						)						)





@torch.jit.ignore
def   _a		(lowercase__			:   Sequence[int]							,   lowercase__			:   Sequence[int]							,   lowercase__			:   Sequence[int]							,   lowercase__			:   Optional[Sequence[bool]] = None							,   lowercase__			:   Optional[Sequence[bool]] = None							,   )			->			List[Tuple[slice, ...]]:






     """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(lowercase__			:   List[bool]						) -> None:
          __snake_case           =						True
          for i in range(len(lowercase__						)						):
               __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(lowercase__						)
     if end_edges is None:
          __snake_case           =						[e == (d - 1) for e, d in zip(lowercase__							,   lowercase__						)]
          reduce_edge_list(lowercase__						)

     # Base cases. Either start/end are empty and we're done, or the final,
     # one-dimensional tensor can be simply sliced
     if len(lowercase__						) == 0:
          return [()]
     elif len(lowercase__						) == 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(lowercase__							,   lowercase__						):
          if s == e:
               path_list.append(slice(lowercase__							,   s + 1						)						)
          else:
               break

     __snake_case           =						tuple(lowercase__						)
     __snake_case           =						len(lowercase__						)

     # start == end, and we're done
     if divergence_idx == len(lowercase__						):
          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(lowercase__							,   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(lowercase__							,   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		(lowercase__			:   torch.Tensor							,   lowercase__			:   int							,   lowercase__			:   int							,   lowercase__			:   int						)			->			torch.Tensor:






     """simple docstring"""
     __snake_case           =						t.shape[:no_batch_dims]
     __snake_case           =						list(_flat_idx_to_idx(lowercase__							,   lowercase__						)						)
     # _get_minimal_slice_set is inclusive
     __snake_case           =						list(_flat_idx_to_idx(flat_end - 1							,   lowercase__						)						)

     # Get an ordered list of slices to perform
     __snake_case           =						_get_minimal_slice_set(
         lowercase__							,   lowercase__							,   lowercase__							,   )

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






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

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

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

          __snake_case           =						tensor_tree_map(lowercase__							,   lowercase__						)

          # Run the layer on the chunk
          __snake_case           =						layer(**lowercase__						)

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

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

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

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

     return out




class 			_lowercase :
   def __init__(						self       :      str       ,   SCREAMING_SNAKE_CASE_       :      int = 512       ,   ) ->							Union[str, Any]:
        __snake_case           =						max_chunk_size
        __snake_case           =						None
        __snake_case           =						None
   def a  (						self       :      List[Any]       ,   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

        __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(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

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

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

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

        assert self.cached_chunk_size is not None

        return self.cached_chunk_size










'''simple docstring'''



from collections import defaultdict

from graphs.minimum_spanning_tree_prims import prisms_algorithm as mst




def   lowercase__(			):

      """simple docstring"""






      SCREAMING_SNAKE_CASE					,    SCREAMING_SNAKE_CASE   :      str    					=						9, 14  # noqa: F841
      SCREAMING_SNAKE_CASE   :      Optional[Any]    					=						[
          [0, 1, 4],
          [0, 7, 8],
          [1, 2, 8],
          [7, 8, 7],
          [7, 6, 1],
          [2, 8, 2],
          [8, 6, 6],
          [2, 3, 7],
          [2, 5, 4],
          [6, 5, 2],
          [3, 5, 14],
          [3, 4, 9],
          [5, 4, 10],
          [1, 7, 11],
      ]

      SCREAMING_SNAKE_CASE   :      Optional[int]    					=						defaultdict(__UpperCamelCase    )
      for nodea, nodea, cost in edges:
            adjancency[nodea].append([nodea, cost]    )
            adjancency[nodea].append([nodea, cost]    )

      SCREAMING_SNAKE_CASE   :      Dict    					=						mst(__UpperCamelCase    )

      SCREAMING_SNAKE_CASE   :      Optional[int]    					=						[
          [7, 6, 1],
          [2, 8, 2],
          [6, 5, 2],
          [0, 1, 4],
          [2, 5, 4],
          [2, 3, 7],
          [0, 7, 8],
          [3, 4, 9],
      ]

      for answer in expected:
            SCREAMING_SNAKE_CASE   :      Any    					=						tuple(answer[:2]    )
            SCREAMING_SNAKE_CASE   :      List[Any]    					=						tuple(edge[::-1]    )
            assert edge in result or reverse in result


