






"""simple docstring"""
from dataclasses import dataclass
from typing import List, Optional, Tuple, Union

import numpy as np
import torch

from ..configuration_utils import ConfigMixin, register_to_config
from ..utils import BaseOutput, logging, randn_tensor
from .scheduling_utils import SchedulerMixin


lowerCAmelCase__							= logging.get_logger(__name__)  # pylint: disable=invalid-name


@dataclass
class _lowerCAmelCase	(						__UpperCAmelCase					):
						SCREAMING_SNAKE_CASE_:				torch.FloatTensor







class _lowerCAmelCase	(						__UpperCAmelCase      ,     __UpperCAmelCase					):
						SCREAMING_SNAKE_CASE_:				List[str]   				=							1







						@register_to_config
						def __init__(						self      ,							lowerCAmelCase_ = 4_0      ,							lowerCAmelCase_ = 0.002      ,							lowerCAmelCase_ = 80.0      ,							lowerCAmelCase_ = 0.5      ,							lowerCAmelCase_ = 1.0      ,							lowerCAmelCase_ = 7.0      ,							lowerCAmelCase_ = True      ,							)		->       Optional[int]:
										# standard deviation of the initial noise distribution
										_SCREAMING_SNAKE_CASE      :		Tuple							=						sigma_max

										_SCREAMING_SNAKE_CASE      :		List[str]							=						np.linspace(0      ,							1      ,							lowerCAmelCase_       )
										_SCREAMING_SNAKE_CASE      :		List[Any]							=						self._convert_to_karras(lowerCAmelCase_       )
										_SCREAMING_SNAKE_CASE      :		Optional[int]							=						self.sigma_to_t(lowerCAmelCase_       )

										# setable values
										_SCREAMING_SNAKE_CASE      :		Optional[Any]							=						None
										_SCREAMING_SNAKE_CASE      :		str							=						torch.from_numpy(lowerCAmelCase_       )
										_SCREAMING_SNAKE_CASE      :		Optional[int]							=						torch.from_numpy(lowerCAmelCase_       )
										_SCREAMING_SNAKE_CASE      :		List[str]							=						False
										_SCREAMING_SNAKE_CASE      :		str							=						False







						def  A			(						self      ,							lowerCAmelCase_      ,							lowerCAmelCase_=None       )		->       Union[str, Any]:
										if schedule_timesteps is None:
														_SCREAMING_SNAKE_CASE      :		List[Any]							=						self.timesteps

										_SCREAMING_SNAKE_CASE      :		str							=						(schedule_timesteps == timestep).nonzero()
										return indices.item()







						def  A			(						self      ,							lowerCAmelCase_      ,							lowerCAmelCase_       )		->       torch.FloatTensor:
										# Get sigma corresponding to timestep
										if isinstance(lowerCAmelCase_      ,							torch.Tensor       ):
														_SCREAMING_SNAKE_CASE      :		Dict							=						timestep.to(self.timesteps.device       )
										_SCREAMING_SNAKE_CASE      :		Union[str, Any]							=						self.index_for_timestep(lowerCAmelCase_       )
										_SCREAMING_SNAKE_CASE      :		Tuple							=						self.sigmas[step_idx]

										_SCREAMING_SNAKE_CASE      :		int							=						sample / ((sigma**2 + self.config.sigma_data**2) ** 0.5)

										_SCREAMING_SNAKE_CASE      :		List[str]							=						True
										return sample







						def  A			(						self      ,							lowerCAmelCase_       )		->       Union[str, Any]:
										if not isinstance(lowerCAmelCase_      ,							np.ndarray       ):
														_SCREAMING_SNAKE_CASE      :		str							=						np.array(lowerCAmelCase_      ,							dtype=np.floataa       )

										_SCREAMING_SNAKE_CASE      :		Tuple							=						1_0_0_0 * 0.25 * np.log(sigmas + 1e-44       )

										return timesteps







						def  A			(						self      ,							lowerCAmelCase_ = None      ,							lowerCAmelCase_ = None      ,							lowerCAmelCase_ = None      ,							)		->       Dict:
										if num_inference_steps is None and timesteps is None:
														raise ValueError('Exactly one of `num_inference_steps` or `timesteps` must be supplied.'       )

										if num_inference_steps is not None and timesteps is not None:
														raise ValueError('Can only pass one of `num_inference_steps` or `timesteps`.'       )

										# Follow DDPMScheduler custom timesteps logic
										if timesteps is not None:
														for i in range(1      ,							len(lowerCAmelCase_       )       ):
																		if timesteps[i] >= timesteps[i - 1]:
																						raise ValueError('`timesteps` must be in descending order.'       )

														if timesteps[0] >= self.config.num_train_timesteps:
																		raise ValueError(
																		    F"""`timesteps` must start before `self.config.train_timesteps`:"""
																		    F""" {self.config.num_train_timesteps}."""       )

														_SCREAMING_SNAKE_CASE      :		Optional[Any]							=						np.array(lowerCAmelCase_      ,							dtype=np.intaa       )
														_SCREAMING_SNAKE_CASE      :		List[str]							=						True
										else:
														if num_inference_steps > self.config.num_train_timesteps:
																		raise ValueError(
																		    F"""`num_inference_steps`: {num_inference_steps} cannot be larger than `self.config.train_timesteps`:"""
																		    F""" {self.config.num_train_timesteps} as the unet model trained with this scheduler can only handle"""
																		    F""" maximal {self.config.num_train_timesteps} timesteps."""       )

														_SCREAMING_SNAKE_CASE      :		List[Any]							=						num_inference_steps

														_SCREAMING_SNAKE_CASE      :		Union[str, Any]							=						self.config.num_train_timesteps // self.num_inference_steps
														_SCREAMING_SNAKE_CASE      :		str							=						(np.arange(0      ,							lowerCAmelCase_       ) * step_ratio).round()[::-1].copy().astype(np.intaa       )
														_SCREAMING_SNAKE_CASE      :		int							=						False

										# Map timesteps to Karras sigmas directly for multistep sampling
										# See https://github.com/openai/consistency_models/blob/main/cm/karras_diffusion.py#L675
										_SCREAMING_SNAKE_CASE      :		Optional[Any]							=						self.config.num_train_timesteps
										_SCREAMING_SNAKE_CASE      :		str							=						timesteps[::-1].copy()
										_SCREAMING_SNAKE_CASE      :		Tuple							=						ramp / (num_train_timesteps - 1)
										_SCREAMING_SNAKE_CASE      :		Any							=						self._convert_to_karras(lowerCAmelCase_       )
										_SCREAMING_SNAKE_CASE      :		Dict							=						self.sigma_to_t(lowerCAmelCase_       )

										_SCREAMING_SNAKE_CASE      :		Union[str, Any]							=						np.concatenate([sigmas, [self.sigma_min]]       ).astype(np.floataa       )
										_SCREAMING_SNAKE_CASE      :		List[Any]							=						torch.from_numpy(lowerCAmelCase_       ).to(device=lowerCAmelCase_       )

										if str(lowerCAmelCase_       ).startswith('mps'       ):
														# mps does not support float64
														_SCREAMING_SNAKE_CASE      :		Dict							=						torch.from_numpy(lowerCAmelCase_       ).to(lowerCAmelCase_      ,							dtype=torch.floataa       )
										else:
														_SCREAMING_SNAKE_CASE      :		Union[str, Any]							=						torch.from_numpy(lowerCAmelCase_       ).to(device=lowerCAmelCase_       )







						def  A			(						self      ,							lowerCAmelCase_       )		->       int:
										_SCREAMING_SNAKE_CASE      :		float							=						self.config.sigma_min
										_SCREAMING_SNAKE_CASE      :		float							=						self.config.sigma_max

										_SCREAMING_SNAKE_CASE      :		Any							=						self.config.rho
										_SCREAMING_SNAKE_CASE      :		str							=						sigma_min ** (1 / rho)
										_SCREAMING_SNAKE_CASE      :		Any							=						sigma_max ** (1 / rho)
										_SCREAMING_SNAKE_CASE      :		Tuple							=						(max_inv_rho + ramp * (min_inv_rho - max_inv_rho)) ** rho
										return sigmas







						def  A			(						self      ,							lowerCAmelCase_       )		->       Tuple:
										_SCREAMING_SNAKE_CASE      :		Any							=						self.config.sigma_data

										_SCREAMING_SNAKE_CASE      :		Tuple							=						sigma_data**2 / (sigma**2 + sigma_data**2)
										_SCREAMING_SNAKE_CASE      :		int							=						sigma * sigma_data / (sigma**2 + sigma_data**2) ** 0.5
										return c_skip, c_out







						def  A			(						self      ,							lowerCAmelCase_       )		->       Any:
										_SCREAMING_SNAKE_CASE      :		List[Any]							=						self.config.sigma_min
										_SCREAMING_SNAKE_CASE      :		List[Any]							=						self.config.sigma_data

										_SCREAMING_SNAKE_CASE      :		Dict							=						sigma_data**2 / ((sigma - sigma_min) ** 2 + sigma_data**2)
										_SCREAMING_SNAKE_CASE      :		Optional[Any]							=						(sigma - sigma_min) * sigma_data / (sigma**2 + sigma_data**2) ** 0.5
										return c_skip, c_out







						def  A			(						self      ,							lowerCAmelCase_      ,							lowerCAmelCase_      ,							lowerCAmelCase_      ,							lowerCAmelCase_ = None      ,							lowerCAmelCase_ = True      ,							)		->       Union[CMStochasticIterativeSchedulerOutput, Tuple]:

										if (
										    isinstance(lowerCAmelCase_      ,							lowerCAmelCase_       )
										    or isinstance(lowerCAmelCase_      ,							torch.IntTensor       )
										    or isinstance(lowerCAmelCase_      ,							torch.LongTensor       )
										):
														raise ValueError(
														    (
														        'Passing integer indices (e.g. from `enumerate(timesteps)`) as timesteps to'
														        F""" `{self.__class__}.step()` is not supported. Make sure to pass"""
														        ' one of the `scheduler.timesteps` as a timestep.'
														    )      ,							)

										if not self.is_scale_input_called:
														logger.warning(
														    'The `scale_model_input` function should be called before `step` to ensure correct denoising. '
														    'See `StableDiffusionPipeline` for a usage example.'       )

										if isinstance(lowerCAmelCase_      ,							torch.Tensor       ):
														_SCREAMING_SNAKE_CASE      :		Dict							=						timestep.to(self.timesteps.device       )

										_SCREAMING_SNAKE_CASE      :		Optional[Any]							=						self.config.sigma_min
										_SCREAMING_SNAKE_CASE      :		Tuple							=						self.config.sigma_max

										_SCREAMING_SNAKE_CASE      :		Optional[int]							=						self.index_for_timestep(lowerCAmelCase_       )

										# sigma_next corresponds to next_t in original implementation
										_SCREAMING_SNAKE_CASE      :		str							=						self.sigmas[step_index]
										if step_index + 1 < self.config.num_train_timesteps:
														_SCREAMING_SNAKE_CASE      :		List[str]							=						self.sigmas[step_index + 1]
										else:
														# Set sigma_next to sigma_min
														_SCREAMING_SNAKE_CASE      :		Optional[Any]							=						self.sigmas[-1]

										# Get scalings for boundary conditions
										_SCREAMING_SNAKE_CASE		,	_SCREAMING_SNAKE_CASE      :		Optional[Any]							=						self.get_scalings_for_boundary_condition(lowerCAmelCase_       )

										# 1. Denoise model output using boundary conditions
										_SCREAMING_SNAKE_CASE      :		List[Any]							=						c_out * model_output + c_skip * sample
										if self.config.clip_denoised:
														_SCREAMING_SNAKE_CASE      :		Union[str, Any]							=						denoised.clamp(-1      ,							1       )

										# 2. Sample z ~ N(0, s_noise^2 * I)
										# Noise is not used for onestep sampling.
										if len(self.timesteps       ) > 1:
														_SCREAMING_SNAKE_CASE      :		int							=						randn_tensor(
														    model_output.shape      ,							dtype=model_output.dtype      ,							device=model_output.device      ,							generator=lowerCAmelCase_       )
										else:
														_SCREAMING_SNAKE_CASE      :		List[Any]							=						torch.zeros_like(lowerCAmelCase_       )
										_SCREAMING_SNAKE_CASE      :		List[str]							=						noise * self.config.s_noise

										_SCREAMING_SNAKE_CASE      :		Union[str, Any]							=						sigma_next.clamp(min=lowerCAmelCase_      ,							max=lowerCAmelCase_       )

										# 3. Return noisy sample
										# tau = sigma_hat, eps = sigma_min
										_SCREAMING_SNAKE_CASE      :		Optional[int]							=						denoised + z * (sigma_hat**2 - sigma_min**2) ** 0.5

										if not return_dict:
														return (prev_sample,)

										return CMStochasticIterativeSchedulerOutput(prev_sample=lowerCAmelCase_       )







						def  A			(						self      ,							lowerCAmelCase_      ,							lowerCAmelCase_      ,							lowerCAmelCase_      ,							)		->       torch.FloatTensor:
										# Make sure sigmas and timesteps have the same device and dtype as original_samples
										_SCREAMING_SNAKE_CASE      :		Dict							=						self.sigmas.to(device=original_samples.device      ,							dtype=original_samples.dtype       )
										if original_samples.device.type == "mps" and torch.is_floating_point(lowerCAmelCase_       ):
														# mps does not support float64
														_SCREAMING_SNAKE_CASE      :		int							=						self.timesteps.to(original_samples.device      ,							dtype=torch.floataa       )
														_SCREAMING_SNAKE_CASE      :		Optional[Any]							=						timesteps.to(original_samples.device      ,							dtype=torch.floataa       )
										else:
														_SCREAMING_SNAKE_CASE      :		Optional[int]							=						self.timesteps.to(original_samples.device       )
														_SCREAMING_SNAKE_CASE      :		List[str]							=						timesteps.to(original_samples.device       )

										_SCREAMING_SNAKE_CASE      :		Tuple							=						[(schedule_timesteps == t).nonzero().item() for t in timesteps]

										_SCREAMING_SNAKE_CASE      :		int							=						sigmas[step_indices].flatten()
										while len(sigma.shape       ) < len(original_samples.shape       ):
														_SCREAMING_SNAKE_CASE      :		Union[str, Any]							=						sigma.unsqueeze(-1       )

										_SCREAMING_SNAKE_CASE      :		List[str]							=						original_samples + noise * sigma
										return noisy_samples







						def __len__(						self       )		->       int:
										return self.config.num_train_timesteps









"""simple docstring"""
import gc
import unittest

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

from diffusers import (
    AutoencoderKL,
    DDIMParallelScheduler,
    DDPMParallelScheduler,
    StableDiffusionParadigmsPipeline,
    UNetaDConditionModel,
)
from diffusers.utils import slow, torch_device
from diffusers.utils.testing_utils import (
    enable_full_determinism,
    require_torch_gpu,
)

from ..pipeline_params import TEXT_TO_IMAGE_BATCH_PARAMS, TEXT_TO_IMAGE_IMAGE_PARAMS, TEXT_TO_IMAGE_PARAMS
from ..test_pipelines_common import PipelineLatentTesterMixin, PipelineTesterMixin


enable_full_determinism()


class _lowerCAmelCase	(						__UpperCAmelCase      ,     __UpperCAmelCase      ,     unittest.TestCase					):
						SCREAMING_SNAKE_CASE_:				int   				=							StableDiffusionParadigmsPipeline
						SCREAMING_SNAKE_CASE_:				str   				=							TEXT_TO_IMAGE_PARAMS
						SCREAMING_SNAKE_CASE_:				Any   				=							TEXT_TO_IMAGE_BATCH_PARAMS
						SCREAMING_SNAKE_CASE_:				str   				=							TEXT_TO_IMAGE_IMAGE_PARAMS
						SCREAMING_SNAKE_CASE_:				List[Any]   				=							TEXT_TO_IMAGE_IMAGE_PARAMS







						def  A			(						self       )		->       int:
										torch.manual_seed(0       )
										_SCREAMING_SNAKE_CASE      :		Dict							=						UNetaDConditionModel(
										    block_out_channels=(3_2, 6_4)      ,							layers_per_block=2      ,							sample_size=3_2      ,							in_channels=4      ,							out_channels=4      ,							down_block_types=('DownBlock2D', 'CrossAttnDownBlock2D')      ,							up_block_types=('CrossAttnUpBlock2D', 'UpBlock2D')      ,							cross_attention_dim=3_2      ,							attention_head_dim=(2, 4)      ,							use_linear_projection=lowerCAmelCase_      ,							)
										_SCREAMING_SNAKE_CASE      :		Union[str, Any]							=						DDIMParallelScheduler(
										    beta_start=0.00_085      ,							beta_end=0.012      ,							beta_schedule='scaled_linear'      ,							clip_sample=lowerCAmelCase_      ,							set_alpha_to_one=lowerCAmelCase_      ,							)
										torch.manual_seed(0       )
										_SCREAMING_SNAKE_CASE      :		List[Any]							=						AutoencoderKL(
										    block_out_channels=[3_2, 6_4]      ,							in_channels=3      ,							out_channels=3      ,							down_block_types=['DownEncoderBlock2D', 'DownEncoderBlock2D']      ,							up_block_types=['UpDecoderBlock2D', 'UpDecoderBlock2D']      ,							latent_channels=4      ,							sample_size=1_2_8      ,							)
										torch.manual_seed(0       )
										_SCREAMING_SNAKE_CASE      :		Optional[int]							=						CLIPTextConfig(
										    bos_token_id=0      ,							eos_token_id=2      ,							hidden_size=3_2      ,							intermediate_size=3_7      ,							layer_norm_eps=1e-05      ,							num_attention_heads=4      ,							num_hidden_layers=5      ,							pad_token_id=1      ,							vocab_size=1_0_0_0      ,							hidden_act='gelu'      ,							projection_dim=5_1_2      ,							)
										_SCREAMING_SNAKE_CASE      :		Union[str, Any]							=						CLIPTextModel(lowerCAmelCase_       )
										_SCREAMING_SNAKE_CASE      :		Optional[Any]							=						CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip'       )

										_SCREAMING_SNAKE_CASE      :		List[Any]							=						{
										    'unet': unet,
										    'scheduler': scheduler,
										    'vae': vae,
										    'text_encoder': text_encoder,
										    'tokenizer': tokenizer,
										    'safety_checker': None,
										    'feature_extractor': None,
										}
										return components







						def  A			(						self      ,							lowerCAmelCase_      ,							lowerCAmelCase_=0       )		->       Tuple:
										if str(lowerCAmelCase_       ).startswith('mps'       ):
														_SCREAMING_SNAKE_CASE      :		Union[str, Any]							=						torch.manual_seed(lowerCAmelCase_       )
										else:
														_SCREAMING_SNAKE_CASE      :		int							=						torch.Generator(device=lowerCAmelCase_       ).manual_seed(lowerCAmelCase_       )
										_SCREAMING_SNAKE_CASE      :		str							=						{
										    'prompt': 'a photograph of an astronaut riding a horse',
										    'generator': generator,
										    'num_inference_steps': 1_0,
										    'guidance_scale': 6.0,
										    'output_type': 'numpy',
										    'parallel': 3,
										    'debug': True,
										}
										return inputs







						def  A			(						self       )		->       int:
										_SCREAMING_SNAKE_CASE      :		Union[str, Any]							=						'cpu'  # ensure determinism for the device-dependent torch.Generator
										_SCREAMING_SNAKE_CASE      :		Optional[int]							=						self.get_dummy_components()
										_SCREAMING_SNAKE_CASE      :		Any							=						StableDiffusionParadigmsPipeline(**lowerCAmelCase_       )
										_SCREAMING_SNAKE_CASE      :		Union[str, Any]							=						sd_pipe.to(lowerCAmelCase_       )
										sd_pipe.set_progress_bar_config(disable=lowerCAmelCase_       )

										_SCREAMING_SNAKE_CASE      :		Optional[int]							=						self.get_dummy_inputs(lowerCAmelCase_       )
										_SCREAMING_SNAKE_CASE      :		Dict							=						sd_pipe(**lowerCAmelCase_       ).images
										_SCREAMING_SNAKE_CASE      :		Dict							=						image[0, -3:, -3:, -1]
										assert image.shape == (1, 6_4, 6_4, 3)

										_SCREAMING_SNAKE_CASE      :		int							=						np.array([0.4_773, 0.5_417, 0.4_723, 0.4_925, 0.5_631, 0.4_752, 0.5_240, 0.4_935, 0.5_023]       )

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







						def  A			(						self       )		->       Tuple:
										_SCREAMING_SNAKE_CASE      :		Tuple							=						'cpu'  # ensure determinism for the device-dependent torch.Generator
										_SCREAMING_SNAKE_CASE      :		str							=						self.get_dummy_components()
										torch.manual_seed(0       )
										_SCREAMING_SNAKE_CASE      :		List[Any]							=						DDPMParallelScheduler()
										torch.manual_seed(0       )
										_SCREAMING_SNAKE_CASE      :		Optional[int]							=						StableDiffusionParadigmsPipeline(**lowerCAmelCase_       )
										_SCREAMING_SNAKE_CASE      :		int							=						sd_pipe.to(lowerCAmelCase_       )
										sd_pipe.set_progress_bar_config(disable=lowerCAmelCase_       )

										_SCREAMING_SNAKE_CASE      :		Tuple							=						self.get_dummy_inputs(lowerCAmelCase_       )
										_SCREAMING_SNAKE_CASE      :		List[Any]							=						sd_pipe(**lowerCAmelCase_       ).images
										_SCREAMING_SNAKE_CASE      :		Dict							=						image[0, -3:, -3:, -1]
										assert image.shape == (1, 6_4, 6_4, 3)

										_SCREAMING_SNAKE_CASE      :		Dict							=						np.array([0.3_573, 0.4_420, 0.4_960, 0.4_799, 0.3_796, 0.3_879, 0.4_819, 0.4_365, 0.4_468]       )

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







						def  A			(						self       )		->       Optional[Any]:
										super().test_inference_batch_consistent(batch_sizes=[1, 2]       )







						def  A			(						self       )		->       Tuple:
										super().test_inference_batch_single_identical(batch_size=2      ,							expected_max_diff=3e-3       )







						def  A			(						self       )		->       List[str]:
										_SCREAMING_SNAKE_CASE      :		List[Any]							=						'cpu'  # ensure determinism for the device-dependent torch.Generator
										_SCREAMING_SNAKE_CASE      :		Optional[Any]							=						self.get_dummy_components()
										_SCREAMING_SNAKE_CASE      :		Union[str, Any]							=						StableDiffusionParadigmsPipeline(**lowerCAmelCase_       )
										_SCREAMING_SNAKE_CASE      :		List[str]							=						sd_pipe.to(lowerCAmelCase_       )
										sd_pipe.set_progress_bar_config(disable=lowerCAmelCase_       )

										_SCREAMING_SNAKE_CASE      :		int							=						self.get_dummy_inputs(lowerCAmelCase_       )
										_SCREAMING_SNAKE_CASE      :		Tuple							=						'french fries'
										_SCREAMING_SNAKE_CASE      :		Optional[Any]							=						sd_pipe(**lowerCAmelCase_      ,							negative_prompt=lowerCAmelCase_       )
										_SCREAMING_SNAKE_CASE      :		Optional[Any]							=						output.images
										_SCREAMING_SNAKE_CASE      :		Dict							=						image[0, -3:, -3:, -1]

										assert image.shape == (1, 6_4, 6_4, 3)

										_SCREAMING_SNAKE_CASE      :		Optional[int]							=						np.array([0.4_771, 0.5_420, 0.4_683, 0.4_918, 0.5_636, 0.4_725, 0.5_230, 0.4_923, 0.5_015]       )

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







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







						def  A			(						self       )		->       Tuple:
										super().tearDown()
										gc.collect()
										torch.cuda.empty_cache()







						def  A			(						self      ,							lowerCAmelCase_=0       )		->       int:
										_SCREAMING_SNAKE_CASE      :		str							=						torch.Generator(device=lowerCAmelCase_       ).manual_seed(lowerCAmelCase_       )
										_SCREAMING_SNAKE_CASE      :		List[str]							=						{
										    'prompt': 'a photograph of an astronaut riding a horse',
										    'generator': generator,
										    'num_inference_steps': 1_0,
										    'guidance_scale': 7.5,
										    'output_type': 'numpy',
										    'parallel': 3,
										    'debug': True,
										}
										return inputs







						def  A			(						self       )		->       List[str]:
										_SCREAMING_SNAKE_CASE      :		Any							=						'stabilityai/stable-diffusion-2-base'
										_SCREAMING_SNAKE_CASE      :		Dict							=						DDIMParallelScheduler.from_pretrained(lowerCAmelCase_      ,							subfolder='scheduler'       )
										_SCREAMING_SNAKE_CASE      :		List[str]							=						StableDiffusionParadigmsPipeline.from_pretrained(lowerCAmelCase_      ,							scheduler=lowerCAmelCase_      ,							safety_checker=lowerCAmelCase_       )
										pipe.to(lowerCAmelCase_       )
										pipe.set_progress_bar_config(disable=lowerCAmelCase_       )
										pipe.enable_attention_slicing()

										_SCREAMING_SNAKE_CASE      :		Optional[int]							=						self.get_inputs()
										_SCREAMING_SNAKE_CASE      :		str							=						pipe(**lowerCAmelCase_       ).images
										_SCREAMING_SNAKE_CASE      :		Dict							=						image[0, -3:, -3:, -1].flatten()

										assert image.shape == (1, 5_1_2, 5_1_2, 3)

										_SCREAMING_SNAKE_CASE      :		Any							=						np.array([0.9_622, 0.9_602, 0.9_748, 0.9_591, 0.9_630, 0.9_691, 0.9_661, 0.9_631, 0.9_741]       )

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

