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]      ):
 SCREAMING_SNAKE_CASE__:   Any=      []
 if isinstance(lowerCamelCase_       ,				lowerCamelCase_      ):
  for v in tree.values():
   shapes.extend(_fetch_dims(lowerCamelCase_      )      )
 elif isinstance(lowerCamelCase_       ,				(list, tuple)      ):
  for t in tree:
   shapes.extend(_fetch_dims(lowerCamelCase_      )      )
 elif isinstance(lowerCamelCase_       ,				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, ...]      ):
 SCREAMING_SNAKE_CASE__:   Tuple=      []
 for d in reversed(lowerCamelCase_      ):
  idx.append(flat_idx % d      )
  SCREAMING_SNAKE_CASE__:   List[Any]=      flat_idx // d

 return tuple(reversed(lowerCamelCase_      )      )







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

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

 if start_edges is None:
  SCREAMING_SNAKE_CASE__:   Any=      [s == 0 for s in start]
  reduce_edge_list(lowerCamelCase_      )
 if end_edges is None:
  SCREAMING_SNAKE_CASE__:   Tuple=      [e == (d - 1) for e, d in zip(lowerCamelCase_       ,				lowerCamelCase_      )]
  reduce_edge_list(lowerCamelCase_      )

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

 SCREAMING_SNAKE_CASE__:   Optional[Any]=      []
 SCREAMING_SNAKE_CASE__:   Tuple=      []

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

 SCREAMING_SNAKE_CASE__:   Optional[int]=      tuple(lowerCamelCase_      )
 SCREAMING_SNAKE_CASE__:   str=      len(lowerCamelCase_      )

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

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

  SCREAMING_SNAKE_CASE__:   int=      start[divergence_idx]
  return tuple(
      path + (slice(lowerCamelCase_       ,				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

  SCREAMING_SNAKE_CASE__:   str=      end[divergence_idx]
  return tuple(
      path + (slice(lowerCamelCase_       ,				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()      )
  SCREAMING_SNAKE_CASE__:   Tuple=      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      ):
 SCREAMING_SNAKE_CASE__:   Optional[Any]=      t.shape[:no_batch_dims]
 SCREAMING_SNAKE_CASE__:   str=      list(_flat_idx_to_idx(lowerCamelCase_       ,				lowerCamelCase_      )      )
 # _get_minimal_slice_set is inclusive
 SCREAMING_SNAKE_CASE__:   Optional[Any]=      list(_flat_idx_to_idx(flat_end - 1       ,				lowerCamelCase_      )      )

 # Get an ordered list of slices to perform
 SCREAMING_SNAKE_CASE__:   Optional[int]=      _get_minimal_slice_set(
     lowerCamelCase_       ,				lowerCamelCase_       ,				lowerCamelCase_       ,				)

 SCREAMING_SNAKE_CASE__:   Optional[Any]=      [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       ,				):
 if not (len(lowerCamelCase_      ) > 0):
  raise ValueError('''Must provide at least one input'''      )

 SCREAMING_SNAKE_CASE__:   Dict=      [shape[:no_batch_dims] for shape in _fetch_dims(lowerCamelCase_      )]
 SCREAMING_SNAKE_CASE__:   Dict=      tuple([max(lowerCamelCase_      ) for s in zip(*lowerCamelCase_      )]      )

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

 SCREAMING_SNAKE_CASE__:   int=      tensor_tree_map(_prep_inputs       ,				lowerCamelCase_      )
 SCREAMING_SNAKE_CASE__:   int=      None
 if _out is not None:
  SCREAMING_SNAKE_CASE__:   Optional[int]=      tensor_tree_map(lambda snake_case_      : t.view([-1] + list(t.shape[no_batch_dims:]      )      )       ,				_out      )

 SCREAMING_SNAKE_CASE__:   Tuple=      1
 for d in orig_batch_dims:
  flat_batch_dim *= d

 SCREAMING_SNAKE_CASE__:   Dict=      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

 SCREAMING_SNAKE_CASE__:   Union[str, Any]=      0
 SCREAMING_SNAKE_CASE__:   Dict=      prepped_outputs
 for _ in range(lowerCamelCase_      ):
  # Chunk the input
  if not low_mem:
   SCREAMING_SNAKE_CASE__:   str=      _select_chunk
  else:
   SCREAMING_SNAKE_CASE__:   Dict=      partial(
       _chunk_slice       ,				flat_start=lowerCamelCase_       ,				flat_end=min(lowerCamelCase_       ,				i + chunk_size      )       ,				no_batch_dims=len(lowerCamelCase_      )       ,				)

  SCREAMING_SNAKE_CASE__:   List[Any]=      tensor_tree_map(lowerCamelCase_       ,				lowerCamelCase_      )

  # Run the layer on the chunk
  SCREAMING_SNAKE_CASE__:   List[Any]=      layer(**lowerCamelCase_      )

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

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

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

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

  i += chunk_size

 SCREAMING_SNAKE_CASE__:   List[Any]=      tensor_tree_map(lambda snake_case_      : t.view(orig_batch_dims + t.shape[1:]      )       ,				lowerCamelCase_      )

 return out






class 		_lowerCamelCase		:
    def __init__(  self , lowerCAmelCase = 512 , )       ->					List[str]:
     SCREAMING_SNAKE_CASE__:   List[str]=      max_chunk_size
     SCREAMING_SNAKE_CASE__:   int=      None
     SCREAMING_SNAKE_CASE__:   List[str]=      None
    def 						UpperCamelCase_				(  self , lowerCAmelCase , lowerCAmelCase , lowerCAmelCase   )       ->					int:
     logging.info('''Tuning chunk size...'''   )

     if min_chunk_size >= self.max_chunk_size:
      return min_chunk_size

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

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

     SCREAMING_SNAKE_CASE__:   str=      0
     SCREAMING_SNAKE_CASE__:   Union[str, Any]=      len(_lowerCamelCase   ) - 1
     while i > min_viable_chunk_size_index:
      SCREAMING_SNAKE_CASE__:   List[str]=      test_chunk_size(candidates[i]   )
      if not viable:
       SCREAMING_SNAKE_CASE__:   Optional[int]=      (min_viable_chunk_size_index + i) // 2
      else:
       SCREAMING_SNAKE_CASE__:   Union[str, Any]=      i
       SCREAMING_SNAKE_CASE__:   List[str]=      (i + len(_lowerCamelCase   ) - 1) // 2

     return candidates[min_viable_chunk_size_index]
    def 						UpperCamelCase_				(  self , lowerCAmelCase , lowerCAmelCase   )       ->					bool:
     SCREAMING_SNAKE_CASE__:   Optional[int]=      True
     for aa, aa in zip(_lowerCamelCase , _lowerCamelCase   ):
      assert type(_lowerCamelCase   ) == type(_lowerCamelCase   )
      if isinstance(_lowerCamelCase , (list, tuple)   ):
       consistent &= self._compare_arg_caches(_lowerCamelCase , _lowerCamelCase   )
      elif isinstance(_lowerCamelCase , _lowerCamelCase   ):
       SCREAMING_SNAKE_CASE__:   Tuple=      [v for _, v in sorted(aa.items() , key=lambda lowerCAmelCase   : x[0]   )]
       SCREAMING_SNAKE_CASE__:   int=      [v for _, v in sorted(aa.items() , key=lambda lowerCAmelCase   : x[0]   )]
       consistent &= self._compare_arg_caches(_lowerCamelCase , _lowerCamelCase   )
      else:
       consistent &= aa == aa

     return consistent



    def 						UpperCamelCase_				(  self , lowerCAmelCase , lowerCAmelCase , lowerCAmelCase , )       ->					int:
     SCREAMING_SNAKE_CASE__:   str=      True
     SCREAMING_SNAKE_CASE__:   List[str]=      tree_map(lambda lowerCAmelCase   : a.shape if isinstance(_lowerCamelCase , torch.Tensor   ) else a , _lowerCamelCase , _lowerCamelCase   )
     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(_lowerCamelCase   )
      SCREAMING_SNAKE_CASE__:   Tuple=      self._compare_arg_caches(self.cached_arg_data , _lowerCamelCase   )
     else:
      # Otherwise, we can reuse the precomputed value
      SCREAMING_SNAKE_CASE__:   Tuple=      False

     if not consistent:
      SCREAMING_SNAKE_CASE__:   str=      self._determine_favorable_chunk_size(
          _lowerCamelCase , _lowerCamelCase , _lowerCamelCase , )
      SCREAMING_SNAKE_CASE__:   str=      arg_data

     assert self.cached_chunk_size is not None

     return self.cached_chunk_size


# tests directory-specific settings - this file is run automatically
# by pytest before any tests are run

import sys
import warnings
from os.path import abspath, dirname, join


# allow having multiple repository checkouts and not needing to remember to rerun
# 'pip install -e .[dev]' when switching between checkouts and running tests.
lowercase_     :   List[str] 				=				abspath(join(dirname(dirname(__file__)), 'src'))
sys.path.insert(1, git_repo_path)

# silence FutureWarning warnings in tests since often we can't act on them until
# they become normal warnings - i.e. the tests still need to test the current functionality
warnings.simplefilter(action='ignore', category=FutureWarning)







def 			A__      (	snake_case_ :		Optional[int]      ):
 from diffusers.utils.testing_utils import pytest_addoption_shared

 pytest_addoption_shared(snake_case_      )







def 			A__      (	snake_case_ :		Optional[int]      ):
 from diffusers.utils.testing_utils import pytest_terminal_summary_main

 SCREAMING_SNAKE_CASE__:   Optional[int]=      terminalreporter.config.getoption('''--make-reports'''      )
 if make_reports:
  pytest_terminal_summary_main(snake_case_       ,				id=snake_case_      )

