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

       return shapes


@torch.jit.ignore
def     __magic_name__      ( __snake_case							:			int	,					__snake_case							:			Tuple[int, ...]    )					->							Tuple[int, ...]:
       lowercase       :							Dict									=				[]
       for d in reversed(__snake_case    ):
              idx.append(flat_idx % d    )
              lowercase       :							str									=				flat_idx // d

       return tuple(reversed(__snake_case    )    )


@torch.jit.ignore
def     __magic_name__      ( __snake_case							:			Sequence[int]	,					__snake_case							:			Sequence[int]	,					__snake_case							:			Sequence[int]	,					__snake_case							:			Optional[Sequence[bool]] = None	,					__snake_case							:			Optional[Sequence[bool]] = 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(__snake_case							:			List[bool]    ) -> None:
              lowercase       :							Union[str, Any]									=				True
              for i in range(len(__snake_case    )    ):
                     lowercase       :							List[Any]									=				-1 * (i + 1)
                     l[reversed_idx] &= tally
                     lowercase       :							Tuple									=				l[reversed_idx]

       if start_edges is None:
              lowercase       :							Optional[Any]									=				[s == 0 for s in start]
              reduce_edge_list(__snake_case    )
       if end_edges is None:
              lowercase       :							Dict									=				[e == (d - 1) for e, d in zip(__snake_case	,					__snake_case    )]
              reduce_edge_list(__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(__snake_case    ) == 0:
              return [()]
       elif len(__snake_case    ) == 1:
              return [(slice(start[0]	,					end[0] + 1    ),)]

       lowercase       :							List[Tuple[slice, ...]]									=				[]
       lowercase       :							List[slice]									=				[]

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

       lowercase       :							Tuple[slice, ...]									=				tuple(__snake_case    )
       lowercase       :							Optional[Any]									=				len(__snake_case    )

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

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

              lowercase       :							List[str]									=				start[divergence_idx]
              return tuple(
                  path + (slice(__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

              lowercase       :							Optional[Any]									=				end[divergence_idx]
              return tuple(
                  path + (slice(__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()    )
              lowercase       :							str									=				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     __magic_name__      ( __snake_case							:			torch.Tensor	,					__snake_case							:			int	,					__snake_case							:			int	,					__snake_case							:			int    )					->							torch.Tensor:
       lowercase       :							List[Any]									=				t.shape[:no_batch_dims]
       lowercase       :							Dict									=				list(_flat_idx_to_idx(__snake_case	,					__snake_case    )    )
       # _get_minimal_slice_set is inclusive
       lowercase       :							Optional[int]									=				list(_flat_idx_to_idx(flat_end - 1	,					__snake_case    )    )

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

       lowercase       :							Optional[int]									=				[t[s] for s in slices]

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


def     __magic_name__      ( __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	,					)					->							Any:
       if not (len(__snake_case    ) > 0):
              raise ValueError("Must provide at least one input"    )

       lowercase       :							List[str]									=				[shape[:no_batch_dims] for shape in _fetch_dims(__snake_case    )]
       lowercase       :							Any									=				tuple([max(__snake_case    ) for s in zip(*__snake_case    )]    )

       def _prep_inputs(__snake_case							:			torch.Tensor    ) -> torch.Tensor:
              if not low_mem:
                     if not sum(t.shape[:no_batch_dims]    ) == no_batch_dims:
                            lowercase       :							Optional[int]									=				t.expand(orig_batch_dims + t.shape[no_batch_dims:]    )
                     lowercase       :							List[Any]									=				t.reshape(-1	,					*t.shape[no_batch_dims:]    )
              else:
                     lowercase       :							List[str]									=				t.expand(orig_batch_dims + t.shape[no_batch_dims:]    )
              return t

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

       lowercase       :							Optional[Any]									=				1
       for d in orig_batch_dims:
              flat_batch_dim *= d

       lowercase       :							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

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

              lowercase       :							Dict[str, Any]									=				tensor_tree_map(__snake_case	,					__snake_case    )

              # Run the layer on the chunk
              lowercase       :							str									=				layer(**__snake_case    )

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

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

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

                     assign(__snake_case	,					__snake_case    )
              elif isinstance(__snake_case	,					__snake_case    ):
                     for xa, xa in zip(__snake_case	,					__snake_case    ):
                            if _add_into_out:
                                   xa[i : i + chunk_size] += xa
                            else:
                                   lowercase       :							Tuple									=				xa
              elif isinstance(__snake_case	,					torch.Tensor    ):
                     if _add_into_out:
                            out[i : i + chunk_size] += output_chunk
                     else:
                            lowercase       :							Optional[int]									=				output_chunk
              else:
                     raise ValueError("Not supported"    )

              i += chunk_size

       lowercase       :							Tuple									=				tensor_tree_map(lambda __snake_case    : t.view(orig_batch_dims + t.shape[1:]    )	,					__snake_case    )

       return out

class 						a__		:
   def __init__(     self      ,       _a = 512      ,       ):
          lowercase       :							Tuple									=				max_chunk_size
          lowercase       :							Optional[int]									=				None
          lowercase       :							Optional[tuple]									=				None
   def      __magic_name__     (     self      ,       _a      ,       _a      ,       _a							):
          logging.info("Tuning chunk size..."							)

          if min_chunk_size >= self.max_chunk_size:
                 return min_chunk_size

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

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

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

          return candidates[min_viable_chunk_size_index]
   def      __magic_name__     (     self      ,       _a      ,       _a							):
          lowercase       :							List[str]									=				True
          for aa, aa in zip(_a      ,       _a							):
                 assert type(_a							) == type(_a							)
                 if isinstance(_a      ,       (list, tuple)							):
                        consistent &= self._compare_arg_caches(_a      ,       _a							)
                 elif isinstance(_a      ,       _a							):
                        lowercase       :							Any									=				[v for _, v in sorted(aa.items()      ,       key=lambda _a							: x[0]							)]
                        lowercase       :							Dict									=				[v for _, v in sorted(aa.items()      ,       key=lambda _a							: x[0]							)]
                        consistent &= self._compare_arg_caches(_a      ,       _a							)
                 else:
                        consistent &= aa == aa

          return consistent

   def      __magic_name__     (     self      ,       _a      ,       _a      ,       _a      ,       ):
          lowercase       :							int									=				True
          lowercase       :							tuple									=				tree_map(lambda _a							: a.shape if isinstance(_a      ,       torch.Tensor							) else a      ,       _a      ,       _a							)
          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(_a							)
                 lowercase       :							Optional[int]									=				self._compare_arg_caches(self.cached_arg_data      ,       _a							)
          else:
                 # Otherwise, we can reuse the precomputed value
                 lowercase       :							Union[str, Any]									=				False

          if not consistent:
                 lowercase       :							Optional[int]									=				self._determine_favorable_chunk_size(
                     _a      ,       _a      ,       _a      ,       )
                 lowercase       :							Union[str, Any]									=				arg_data

          assert self.cached_chunk_size is not None

          return self.cached_chunk_size




"""simple docstring"""




import gc
import random
import unittest

import numpy as np
import torch
from transformers import CLIPTextConfig, CLIPTextModel, CLIPTextModelWithProjection, CLIPTokenizer

from diffusers import (
    AutoencoderKL,
    DiffusionPipeline,
    EulerDiscreteScheduler,
    StableDiffusionXLImgaImgPipeline,
    UNetaDConditionModel,
)
from diffusers.utils import floats_tensor, slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu

from ..pipeline_params import (
    IMAGE_TO_IMAGE_IMAGE_PARAMS,
    TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS,
    TEXT_GUIDED_IMAGE_VARIATION_PARAMS,
)
from ..test_pipelines_common import PipelineLatentTesterMixin, PipelineTesterMixin


enable_full_determinism()

class 						a__		( a_,     a_,     unittest.TestCase		):
   __lowerCAmelCase				=							StableDiffusionXLImgaImgPipeline
   __lowerCAmelCase				=							TEXT_GUIDED_IMAGE_VARIATION_PARAMS - {"""height""", """width"""}
   __lowerCAmelCase				=							PipelineTesterMixin.required_optional_params - {"""latents"""}
   __lowerCAmelCase				=							TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS
   __lowerCAmelCase				=							IMAGE_TO_IMAGE_IMAGE_PARAMS
   __lowerCAmelCase				=							IMAGE_TO_IMAGE_IMAGE_PARAMS
   def      __magic_name__     (     self							):
          torch.manual_seed(0							)
          lowercase       :							str									=				UNetaDConditionModel(
              block_out_channels=(32, 64)      ,       layers_per_block=2      ,       sample_size=32      ,       in_channels=4      ,       out_channels=4      ,       down_block_types=("DownBlock2D", "CrossAttnDownBlock2D")      ,       up_block_types=("CrossAttnUpBlock2D", "UpBlock2D")      ,       attention_head_dim=(2, 4)      ,       use_linear_projection=_a      ,       addition_embed_type="text_time"      ,       addition_time_embed_dim=8      ,       transformer_layers_per_block=(1, 2)      ,       projection_class_embeddings_input_dim=80      ,       cross_attention_dim=64      ,       )
          lowercase       :							Optional[Any]									=				EulerDiscreteScheduler(
              beta_start=0.0_0_0_8_5      ,       beta_end=0.0_1_2      ,       steps_offset=1      ,       beta_schedule="scaled_linear"      ,       timestep_spacing="leading"      ,       )
          torch.manual_seed(0							)
          lowercase       :							Any									=				AutoencoderKL(
              block_out_channels=[32, 64]      ,       in_channels=3      ,       out_channels=3      ,       down_block_types=["DownEncoderBlock2D", "DownEncoderBlock2D"]      ,       up_block_types=["UpDecoderBlock2D", "UpDecoderBlock2D"]      ,       latent_channels=4      ,       sample_size=128      ,       )
          torch.manual_seed(0							)
          lowercase       :							Dict									=				CLIPTextConfig(
              bos_token_id=0      ,       eos_token_id=2      ,       hidden_size=32      ,       intermediate_size=37      ,       layer_norm_eps=1E-05      ,       num_attention_heads=4      ,       num_hidden_layers=5      ,       pad_token_id=1      ,       vocab_size=1_000      ,       hidden_act="gelu"      ,       projection_dim=32      ,       )
          lowercase       :							str									=				CLIPTextModel(_a							)
          lowercase       :							List[str]									=				CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip"      ,       local_files_only=_a							)

          lowercase       :							Any									=				CLIPTextModelWithProjection(_a							)
          lowercase       :							Union[str, Any]									=				CLIPTokenizer.from_pretrained("hf-internal-testing/tiny-random-clip"      ,       local_files_only=_a							)

          lowercase       :							Any									=				{
              "unet": unet,
              "scheduler": scheduler,
              "vae": vae,
              "text_encoder": text_encoder,
              "tokenizer": tokenizer,
              "text_encoder_2": text_encoder_a,
              "tokenizer_2": tokenizer_a,
              # "safety_checker": None,
              # "feature_extractor": None,
          }
          return components
   def      __magic_name__     (     self      ,       _a      ,       _a=0							):
          lowercase       :							Dict									=				floats_tensor((1, 3, 32, 32)      ,       rng=random.Random(_a							)							).to(_a							)
          lowercase       :							List[Any]									=				image / 2 + 0.5
          if str(_a							).startswith("mps"							):
                 lowercase       :							Optional[int]									=				torch.manual_seed(_a							)
          else:
                 lowercase       :							str									=				torch.Generator(device=_a							).manual_seed(_a							)
          lowercase       :							Optional[int]									=				{
              "prompt": "A painting of a squirrel eating a burger",
              "image": image,
              "generator": generator,
              "num_inference_steps": 2,
              "guidance_scale": 5.0,
              "output_type": "numpy",
              "strength": 0.7_5,
          }
          return inputs
   def      __magic_name__     (     self							):
          lowercase       :							Dict									=				"cpu"  # ensure determinism for the device-dependent torch.Generator
          lowercase       :							Optional[int]									=				self.get_dummy_components()
          lowercase       :							str									=				StableDiffusionXLImgaImgPipeline(**_a							)
          lowercase       :							List[Any]									=				sd_pipe.to(_a							)
          sd_pipe.set_progress_bar_config(disable=_a							)

          lowercase       :							Optional[Any]									=				self.get_dummy_inputs(_a							)
          lowercase       :							Tuple									=				sd_pipe(**_a							).images
          lowercase       :							Dict									=				image[0, -3:, -3:, -1]

          assert image.shape == (1, 32, 32, 3)

          lowercase       :							Tuple									=				np.array([0.4_6_5_6, 0.4_8_4_0, 0.4_4_3_9, 0.6_6_9_8, 0.5_5_7_4, 0.4_5_2_4, 0.5_7_9_9, 0.5_9_4_3, 0.5_1_6_5]							)

          assert np.abs(image_slice.flatten() - expected_slice							).max() < 1E-2
   def      __magic_name__     (     self							):
          super().test_attention_slicing_forward_pass(expected_max_diff=3E-3							)
   def      __magic_name__     (     self							):
          super().test_inference_batch_single_identical(expected_max_diff=3E-3							)
   def      __magic_name__     (     self							):
          pass

   def      __magic_name__     (     self							):
          lowercase       :							Optional[Any]									=				self.get_dummy_components()
          lowercase       :							Optional[Any]									=				StableDiffusionXLImgaImgPipeline(**_a							)
          lowercase       :							Any									=				sd_pipe.to(_a							)
          lowercase       :							Any									=				sd_pipe.to(_a							)
          sd_pipe.set_progress_bar_config(disable=_a							)

          # forward without prompt embeds
          lowercase       :							Union[str, Any]									=				self.get_dummy_inputs(_a							)
          lowercase       :							int									=				3 * ["this is a negative prompt"]
          lowercase       :							Any									=				negative_prompt
          lowercase       :							List[str]									=				3 * [inputs["prompt"]]

          lowercase       :							Optional[int]									=				sd_pipe(**_a							)
          lowercase       :							List[Any]									=				output.images[0, -3:, -3:, -1]

          # forward with prompt embeds
          lowercase       :							Any									=				self.get_dummy_inputs(_a							)
          lowercase       :							List[str]									=				3 * ["this is a negative prompt"]
          lowercase       :							Optional[Any]									=				3 * [inputs.pop("prompt"							)]

          (
              (
              lowercase
          )						,       (
              lowercase
          )						,       (
              lowercase
          )						,       (
              lowercase
          )						,       
          )       :							List[Any]									=				sd_pipe.encode_prompt(_a      ,       negative_prompt=_a							)

          lowercase       :							int									=				sd_pipe(
              **_a      ,       prompt_embeds=_a      ,       negative_prompt_embeds=_a      ,       pooled_prompt_embeds=_a      ,       negative_pooled_prompt_embeds=_a      ,       )
          lowercase       :							Dict									=				output.images[0, -3:, -3:, -1]

          # make sure that it's equal
          assert np.abs(image_slice_a.flatten() - image_slice_a.flatten()							).max() < 1E-4


@slow
@require_torch_gpu
class 						a__		( unittest.TestCase		):
   def      __magic_name__     (     self							):
          super().tearDown()
          gc.collect()
          torch.cuda.empty_cache()
   def      __magic_name__     (     self      ,       _a      ,       _a="cpu"      ,       _a=torch.floataa      ,       _a=0							):
          lowercase       :							Optional[int]									=				torch.Generator(device=_a							).manual_seed(_a							)
          lowercase       :							Any									=				np.random.RandomState(_a							).standard_normal((1, 4, 64, 64)							)
          lowercase       :							List[str]									=				torch.from_numpy(_a							).to(device=_a      ,       dtype=_a							)
          lowercase       :							List[Any]									=				{
              "prompt": "a photograph of an astronaut riding a horse",
              "latents": latents,
              "generator": generator,
              "num_inference_steps": 3,
              "guidance_scale": 7.5,
              "output_type": "numpy",
          }
          return inputs

   def      __magic_name__     (     self							):
          lowercase       :							List[Any]									=				DiffusionPipeline.from_pretrained("stabilityai/stable-diffusion-2-base"							)
          pipe.to(_a							)
          pipe.set_progress_bar_config(disable=_a							)

          lowercase       :							Any									=				self.get_inputs(_a							)
          lowercase       :							int									=				pipe(**_a							).images
          lowercase       :							Any									=				image[0, -3:, -3:, -1].flatten()

          assert image.shape == (1, 512, 512, 3)
          lowercase       :							Optional[Any]									=				np.array([0.4_9_4_9_3, 0.4_7_8_9_6, 0.4_0_7_9_8, 0.5_4_2_1_4, 0.5_3_2_1_2, 0.4_8_2_0_2, 0.4_7_6_5_6, 0.4_6_3_2_9, 0.4_8_5_0_6]							)
          assert np.abs(image_slice - expected_slice							).max() < 7E-3


