


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   _SCREAMING_SNAKE_CASE	( SCREAMING_SNAKE_CASE       ):
  A_      :				List[Any]		     =					[]
  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   _SCREAMING_SNAKE_CASE	( SCREAMING_SNAKE_CASE ,			SCREAMING_SNAKE_CASE       ):
  A_      :				str		     =					[]
  for d in reversed(SCREAMING_SNAKE_CASE       ):
    idx.append(flat_idx % d       )
    A_      :				List[str]		     =					flat_idx // d

  return tuple(reversed(SCREAMING_SNAKE_CASE       )       )


@torch.jit.ignore
def   _SCREAMING_SNAKE_CASE	( SCREAMING_SNAKE_CASE ,			SCREAMING_SNAKE_CASE ,			SCREAMING_SNAKE_CASE ,			SCREAMING_SNAKE_CASE = None ,			SCREAMING_SNAKE_CASE = 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(SCREAMING_SNAKE_CASE       ) -> None:
    A_      :				Dict		     =					True
    for i in range(len(SCREAMING_SNAKE_CASE       )       ):
      A_      :				Optional[Any]		     =					-1 * (i + 1)
      l[reversed_idx] &= tally
      A_      :				Any		     =					l[reversed_idx]

  if start_edges is None:
    A_      :				Tuple		     =					[s == 0 for s in start]
    reduce_edge_list(SCREAMING_SNAKE_CASE       )
  if end_edges is None:
    A_      :				Union[str, Any]		     =					[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_      :				List[Tuple[slice, ...]]		     =					[]
  A_      :				List[slice]		     =					[]

  # 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[slice, ...]		     =					tuple(SCREAMING_SNAKE_CASE       )
  A_      :				Optional[int]		     =					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_      :				List[Any]		     =					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_      :				Union[str, Any]		     =					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_      :				int		     =					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   _SCREAMING_SNAKE_CASE	( SCREAMING_SNAKE_CASE ,			SCREAMING_SNAKE_CASE ,			SCREAMING_SNAKE_CASE ,			SCREAMING_SNAKE_CASE       ):
  A_      :				Tuple		     =					t.shape[:no_batch_dims]
  A_      :				Tuple		     =					list(_flat_idx_to_idx(SCREAMING_SNAKE_CASE ,			SCREAMING_SNAKE_CASE       )       )
  # _get_minimal_slice_set is inclusive
  A_      :				List[str]		     =					list(_flat_idx_to_idx(flat_end - 1 ,			SCREAMING_SNAKE_CASE       )       )

  # Get an ordered list of slices to perform
  A_      :				List[Any]		     =					_get_minimal_slice_set(
      SCREAMING_SNAKE_CASE ,			SCREAMING_SNAKE_CASE ,			SCREAMING_SNAKE_CASE ,			)

  A_      :				Tuple		     =					[t[s] for s in slices]

  return torch.cat([s.view((-1,) + t.shape[no_batch_dims:]       ) for s in sliced_tensors]       )


def   _SCREAMING_SNAKE_CASE	( SCREAMING_SNAKE_CASE ,			SCREAMING_SNAKE_CASE ,			SCREAMING_SNAKE_CASE ,			SCREAMING_SNAKE_CASE ,			SCREAMING_SNAKE_CASE = False ,			SCREAMING_SNAKE_CASE = None ,			SCREAMING_SNAKE_CASE = False ,			):
  if not (len(SCREAMING_SNAKE_CASE       ) > 0):
    raise ValueError('''Must provide at least one input'''       )

  A_      :				int		     =					[shape[:no_batch_dims] for shape in _fetch_dims(SCREAMING_SNAKE_CASE       )]
  A_      :				int		     =					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:
        A_      :				Any		     =					t.expand(orig_batch_dims + t.shape[no_batch_dims:]       )
      A_      :				List[Any]		     =					t.reshape(-1 ,			*t.shape[no_batch_dims:]       )
    else:
      A_      :				Optional[int]		     =					t.expand(orig_batch_dims + t.shape[no_batch_dims:]       )
    return t

  A_      :				Dict[str, Any]		     =					tensor_tree_map(_prep_inputs ,			SCREAMING_SNAKE_CASE       )
  A_      :				Optional[Any]		     =					None
  if _out is not None:
    A_      :				Optional[int]		     =					tensor_tree_map(lambda SCREAMING_SNAKE_CASE       : t.view([-1] + list(t.shape[no_batch_dims:]       )       ) ,			_out       )

  A_      :				Dict		     =					1
  for d in orig_batch_dims:
    flat_batch_dim *= d

  A_      :				Optional[Any]		     =					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

  A_      :				Union[str, Any]		     =					0
  A_      :				Optional[int]		     =					prepped_outputs
  for _ in range(SCREAMING_SNAKE_CASE       ):
    # Chunk the input
    if not low_mem:
      A_      :				Optional[Any]		     =					_select_chunk
    else:
      A_      :				Dict		     =					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_      :				Dict[str, Any]		     =					tensor_tree_map(SCREAMING_SNAKE_CASE ,			SCREAMING_SNAKE_CASE       )

    # Run the layer on the chunk
    A_      :				List[str]		     =					layer(**SCREAMING_SNAKE_CASE       )

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

    i += chunk_size

  A_      :				Optional[Any]		     =					tensor_tree_map(lambda SCREAMING_SNAKE_CASE       : t.view(orig_batch_dims + t.shape[1:]       ) ,			SCREAMING_SNAKE_CASE       )

  return out







class   _lowerCamelCase :




       """simple docstring"""



       def __init__(      self				,       _SCREAMING_SNAKE_CASE = 512				,       )->Any:




         '''simple docstring'''





         A_      :				Any		     =					max_chunk_size
         A_      :				Optional[int]		     =					None
         A_      :				Optional[tuple]		     =					None



       def        _snake_case       (      self				,       _SCREAMING_SNAKE_CASE				,       _SCREAMING_SNAKE_CASE				,       _SCREAMING_SNAKE_CASE		)->int:




         '''simple docstring'''





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

         if min_chunk_size >= self.max_chunk_size:
           return min_chunk_size

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

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

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

         return candidates[min_viable_chunk_size_index]



       def        _snake_case       (      self				,       _SCREAMING_SNAKE_CASE				,       _SCREAMING_SNAKE_CASE		)->bool:




         '''simple docstring'''





         A_      :				List[Any]		     =					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		):
             A_      :				List[Any]		     =					[v for _, v in sorted(aa.items()				,       key=lambda _SCREAMING_SNAKE_CASE		: x[0]		)]
             A_      :				int		     =					[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        _snake_case       (      self				,       _SCREAMING_SNAKE_CASE				,       _SCREAMING_SNAKE_CASE				,       _SCREAMING_SNAKE_CASE				,       )->int:




         '''simple docstring'''





         A_      :				List[Any]		     =					True
         A_      :				tuple		     =					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		)
           A_      :				Tuple		     =					self._compare_arg_caches(self.cached_arg_data				,       _SCREAMING_SNAKE_CASE		)
         else:
           # Otherwise, we can reuse the precomputed value
           A_      :				Union[str, Any]		     =					False

         if not consistent:
           A_      :				List[Any]		     =					self._determine_favorable_chunk_size(
               _SCREAMING_SNAKE_CASE				,       _SCREAMING_SNAKE_CASE				,       _SCREAMING_SNAKE_CASE				,       )
           A_      :				Union[str, Any]		     =					arg_data

         assert self.cached_chunk_size is not None

         return self.cached_chunk_size








import unittest

import numpy as np
import torch

from diffusers import DDIMPipeline, DDIMScheduler, UNetaDModel
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu, slow, torch_device

from ..pipeline_params import UNCONDITIONAL_IMAGE_GENERATION_BATCH_PARAMS, UNCONDITIONAL_IMAGE_GENERATION_PARAMS
from ..test_pipelines_common import PipelineTesterMixin


enable_full_determinism()






class   _lowerCamelCase (					UpperCamelCase					,							unittest.TestCase			):




       """simple docstring"""
       snake_case      =    DDIMPipeline
       snake_case      =    UNCONDITIONAL_IMAGE_GENERATION_PARAMS
       snake_case      =    PipelineTesterMixin.required_optional_params - {
           "num_images_per_prompt",
           "latents",
           "callback",
           "callback_steps",
       }
       snake_case      =    UNCONDITIONAL_IMAGE_GENERATION_BATCH_PARAMS
       snake_case      =    False



       def        _snake_case       (      self		)->List[str]:




         '''simple docstring'''





         torch.manual_seed(0		)
         A_      :				List[str]		     =					UNetaDModel(
             block_out_channels=(32, 64)				,       layers_per_block=2				,       sample_size=32				,       in_channels=3				,       out_channels=3				,       down_block_types=('''DownBlock2D''', '''AttnDownBlock2D''')				,       up_block_types=('''AttnUpBlock2D''', '''UpBlock2D''')				,       )
         A_      :				Optional[Any]		     =					DDIMScheduler()
         A_      :				str		     =					{'''unet''': unet, '''scheduler''': scheduler}
         return components



       def        _snake_case       (      self				,       _SCREAMING_SNAKE_CASE				,       _SCREAMING_SNAKE_CASE=0		)->Optional[Any]:




         '''simple docstring'''





         if str(_SCREAMING_SNAKE_CASE		).startswith('''mps'''		):
           A_      :				Any		     =					torch.manual_seed(_SCREAMING_SNAKE_CASE		)
         else:
           A_      :				Optional[int]		     =					torch.Generator(device=_SCREAMING_SNAKE_CASE		).manual_seed(_SCREAMING_SNAKE_CASE		)
         A_      :				Any		     =					{
             '''batch_size''': 1,
             '''generator''': generator,
             '''num_inference_steps''': 2,
             '''output_type''': '''numpy''',
         }
         return inputs



       def        _snake_case       (      self		)->List[Any]:




         '''simple docstring'''





         A_      :				Optional[int]		     =					'''cpu'''

         A_      :				Dict		     =					self.get_dummy_components()
         A_      :				str		     =					self.pipeline_class(**_SCREAMING_SNAKE_CASE		)
         pipe.to(_SCREAMING_SNAKE_CASE		)
         pipe.set_progress_bar_config(disable=_SCREAMING_SNAKE_CASE		)

         A_      :				str		     =					self.get_dummy_inputs(_SCREAMING_SNAKE_CASE		)
         A_      :				Any		     =					pipe(**_SCREAMING_SNAKE_CASE		).images
         A_      :				int		     =					image[0, -3:, -3:, -1]

         self.assertEqual(image.shape				,       (1, 32, 32, 3)		)
         A_      :				List[Any]		     =					np.array(
             [1.000e00, 5.717e-01, 4.717e-01, 1.000e00, 0.000e00, 1.000e00, 3.000e-04, 0.000e00, 9.000e-04]		)
         A_      :				str		     =					np.abs(image_slice.flatten() - expected_slice		).max()
         self.assertLessEqual(_SCREAMING_SNAKE_CASE				,       1e-3		)



       def        _snake_case       (      self		)->Union[str, Any]:




         '''simple docstring'''





         super().test_dict_tuple_outputs_equivalent(expected_max_difference=3e-3		)



       def        _snake_case       (      self		)->Optional[int]:




         '''simple docstring'''





         super().test_save_load_local(expected_max_difference=3e-3		)



       def        _snake_case       (      self		)->Optional[int]:




         '''simple docstring'''





         super().test_save_load_optional_components(expected_max_difference=3e-3		)







       def        _snake_case       (      self		)->Any:




         '''simple docstring'''





         super().test_inference_batch_single_identical(expected_max_diff=3e-3		)






@slow
@require_torch_gpu
class   _lowerCamelCase (					unittest.TestCase			):




       """simple docstring"""



       def        _snake_case       (      self		)->Union[str, Any]:




         '''simple docstring'''





         A_      :				int		     =					'''google/ddpm-cifar10-32'''

         A_      :				Tuple		     =					UNetaDModel.from_pretrained(_SCREAMING_SNAKE_CASE		)
         A_      :				str		     =					DDIMScheduler()

         A_      :				str		     =					DDIMPipeline(unet=_SCREAMING_SNAKE_CASE				,       scheduler=_SCREAMING_SNAKE_CASE		)
         ddim.to(_SCREAMING_SNAKE_CASE		)
         ddim.set_progress_bar_config(disable=_SCREAMING_SNAKE_CASE		)

         A_      :				Optional[int]		     =					torch.manual_seed(0		)
         A_      :				Any		     =					ddim(generator=_SCREAMING_SNAKE_CASE				,       eta=0.0				,       output_type='''numpy'''		).images

         A_      :				Any		     =					image[0, -3:, -3:, -1]

         assert image.shape == (1, 32, 32, 3)
         A_      :				Any		     =					np.array([0.1_7_2_3, 0.1_6_1_7, 0.1_6_0_0, 0.1_6_2_6, 0.1_4_9_7, 0.1_5_1_3, 0.1_5_0_5, 0.1_4_4_2, 0.1_4_5_3]		)

         assert np.abs(image_slice.flatten() - expected_slice		).max() < 1e-2







       def        _snake_case       (      self		)->List[str]:




         '''simple docstring'''





         A_      :				Tuple		     =					'''google/ddpm-ema-bedroom-256'''

         A_      :				int		     =					UNetaDModel.from_pretrained(_SCREAMING_SNAKE_CASE		)
         A_      :				Any		     =					DDIMScheduler.from_pretrained(_SCREAMING_SNAKE_CASE		)

         A_      :				Optional[Any]		     =					DDIMPipeline(unet=_SCREAMING_SNAKE_CASE				,       scheduler=_SCREAMING_SNAKE_CASE		)
         ddpm.to(_SCREAMING_SNAKE_CASE		)
         ddpm.set_progress_bar_config(disable=_SCREAMING_SNAKE_CASE		)

         A_      :				Dict		     =					torch.manual_seed(0		)
         A_      :				List[str]		     =					ddpm(generator=_SCREAMING_SNAKE_CASE				,       output_type='''numpy'''		).images

         A_      :				Any		     =					image[0, -3:, -3:, -1]

         assert image.shape == (1, 256, 256, 3)
         A_      :				Tuple		     =					np.array([0.0_0_6_0, 0.0_2_0_1, 0.0_3_4_4, 0.0_0_2_4, 0.0_0_1_8, 0.0_0_0_2, 0.0_0_2_2, 0.0_0_0_0, 0.0_0_6_9]		)

         assert np.abs(image_slice.flatten() - expected_slice		).max() < 1e-2




