






"""simple docstring"""



import json
import os
import shutil
import sys
import tempfile
import unittest
from unittest import TestCase
from unittest.mock import patch

import faiss
import numpy as np
from datasets import Dataset

from transformers import BartConfig, BartTokenizer, DPRConfig, DPRQuestionEncoderTokenizer, RagConfig
from transformers.file_utils import is_datasets_available, is_faiss_available, is_psutil_available, is_torch_available
from transformers.integrations import is_ray_available
from transformers.models.bert.tokenization_bert import VOCAB_FILES_NAMES as DPR_VOCAB_FILES_NAMES
from transformers.models.rag.retrieval_rag import CustomHFIndex, RagRetriever
from transformers.models.roberta.tokenization_roberta import VOCAB_FILES_NAMES as BART_VOCAB_FILES_NAMES
from transformers.testing_utils import require_ray


sys.path.append(os.path.join(os.getcwd()))  # noqa: E402 # noqa: E402 # isort:skip

if is_torch_available():
						from distributed_pytorch_retriever import RagPyTorchDistributedRetriever  # noqa: E402 # isort:skip
else:
						__A		:	List[Any]      =		None

if is_ray_available():
						import ray  # noqa: E402 # isort:skip
						from distributed_ray_retriever import RagRayDistributedRetriever, RayRetriever  # noqa: E402 # isort:skip
else:
						__A		:	Dict      =		None
						__A		:	Tuple      =		None
						__A		:	Optional[int]      =		None



def 						lowercase   (    __snake_case    :			List[str]     ):
		if not (is_datasets_available() and is_faiss_available() and is_psutil_available()):
				lowercase_			:			Union[str, Any]								=       unittest.skip('''test requires Datasets, Faiss, psutil'''     )(__snake_case     )
		return test_case






@require_distributed_retrieval
class  _UpperCAmelCase	(			_A							):




					def   A  (    self					:						str       )  ->  List[str]:
							lowercase_			:			str								=       tempfile.mkdtemp()
							lowercase_			:			Tuple								=       8

							# DPR tok
							lowercase_			:			Optional[Any]								=       [
							    '''[UNK]''',
							    '''[CLS]''',
							    '''[SEP]''',
							    '''[PAD]''',
							    '''[MASK]''',
							    '''want''',
							    '''##want''',
							    '''##ed''',
							    '''wa''',
							    '''un''',
							    '''runn''',
							    '''##ing''',
							    ''',''',
							    '''low''',
							    '''lowest''',
							]
							lowercase_			:			str								=       os.path.join(self.tmpdirname   ,						'''dpr_tokenizer'''       )
							os.makedirs(A   ,						exist_ok=A       )
							lowercase_			:			Optional[Any]								=       os.path.join(A   ,						DPR_VOCAB_FILES_NAMES['''vocab_file''']       )
							with open(self.vocab_file   ,						'''w'''   ,						encoding='''utf-8'''       ) as vocab_writer:
									vocab_writer.write(''''''.join([x + '''\n''' for x in vocab_tokens]       )       )

							# BART tok
							lowercase_			:			List[str]								=       [
							    '''l''',
							    '''o''',
							    '''w''',
							    '''e''',
							    '''r''',
							    '''s''',
							    '''t''',
							    '''i''',
							    '''d''',
							    '''n''',
							    '''\u0120''',
							    '''\u0120l''',
							    '''\u0120n''',
							    '''\u0120lo''',
							    '''\u0120low''',
							    '''er''',
							    '''\u0120lowest''',
							    '''\u0120newer''',
							    '''\u0120wider''',
							    '''<unk>''',
							]
							lowercase_			:			Dict								=       dict(zip(A   ,						range(len(A       )       )       )       )
							lowercase_			:			str								=       ['''#version: 0.2''', '''\u0120 l''', '''\u0120l o''', '''\u0120lo w''', '''e r''', '''''']
							lowercase_			:			Dict								=       {'''unk_token''': '''<unk>'''}

							lowercase_			:			Optional[int]								=       os.path.join(self.tmpdirname   ,						'''bart_tokenizer'''       )
							os.makedirs(A   ,						exist_ok=A       )
							lowercase_			:			Union[str, Any]								=       os.path.join(A   ,						BART_VOCAB_FILES_NAMES['''vocab_file''']       )
							lowercase_			:			Dict								=       os.path.join(A   ,						BART_VOCAB_FILES_NAMES['''merges_file''']       )
							with open(self.vocab_file   ,						'''w'''   ,						encoding='''utf-8'''       ) as fp:
									fp.write(json.dumps(A       ) + '''\n'''       )
							with open(self.merges_file   ,						'''w'''   ,						encoding='''utf-8'''       ) as fp:
									fp.write('''\n'''.join(A       )       )




					def   A  (    self					:						Dict       )  ->  DPRQuestionEncoderTokenizer:
							return DPRQuestionEncoderTokenizer.from_pretrained(os.path.join(self.tmpdirname   ,						'''dpr_tokenizer'''       )       )




					def   A  (    self					:						Optional[int]       )  ->  BartTokenizer:
							return BartTokenizer.from_pretrained(os.path.join(self.tmpdirname   ,						'''bart_tokenizer'''       )       )




					def   A  (    self					:						Union[str, Any]       )  ->  Optional[Any]:
							shutil.rmtree(self.tmpdirname       )




					def   A  (    self					:						Any       )  ->  Union[str, Any]:
							lowercase_			:			Tuple								=       Dataset.from_dict(
							    {
							        '''id''': ['''0''', '''1'''],
							        '''text''': ['''foo''', '''bar'''],
							        '''title''': ['''Foo''', '''Bar'''],
							        '''embeddings''': [np.ones(self.retrieval_vector_size       ), 2 * np.ones(self.retrieval_vector_size       )],
							    }       )
							dataset.add_faiss_index('''embeddings'''   ,						string_factory='''Flat'''   ,						metric_type=faiss.METRIC_INNER_PRODUCT       )
							return dataset




					def   A  (    self					:						Union[str, Any]   ,						A					:						bool   ,						A					:						List[str]=1_23_45       )  ->  RagPyTorchDistributedRetriever:
							lowercase_			:			Dict								=       self.get_dummy_dataset()
							lowercase_			:			List[str]								=       RagConfig(
							    retrieval_vector_size=self.retrieval_vector_size   ,						question_encoder=DPRConfig().to_dict()   ,						generator=BartConfig().to_dict()   ,						)
							with patch('''transformers.models.rag.retrieval_rag.load_dataset'''       ) as mock_load_dataset:
									lowercase_			:			Union[str, Any]								=       dataset
									lowercase_			:			List[Any]								=       RagPyTorchDistributedRetriever(
									    A   ,						question_encoder_tokenizer=self.get_dpr_tokenizer()   ,						generator_tokenizer=self.get_bart_tokenizer()   ,						)
									if init_retrieval:
											retriever.init_retrieval(A       )
							return retriever




					def   A  (    self					:						Optional[int]   ,						A					:						bool       )  ->  RagRayDistributedRetriever:
							# Have to run in local mode because sys.path modifications at top of
							# file are not propogated to remote workers.
							# https://stackoverflow.com/questions/54338013/parallel-import-a-python-file-from-sibling-folder
							ray.init(local_mode=A       )
							lowercase_			:			Tuple								=       RagConfig(
							    retrieval_vector_size=self.retrieval_vector_size   ,						question_encoder=DPRConfig().to_dict()   ,						generator=BartConfig().to_dict()   ,						)
							lowercase_			:			Tuple								=       ray.remote(A       )
							lowercase_			:			Optional[int]								=       [remote_cls.remote() for _ in range(1       )]
							with patch('''transformers.models.rag.retrieval_rag.load_dataset'''       ) as mock_load_dataset:
									lowercase_			:			Tuple								=       self.get_dummy_dataset()
									lowercase_			:			Union[str, Any]								=       RagRayDistributedRetriever(
									    A   ,						question_encoder_tokenizer=self.get_dpr_tokenizer()   ,						generator_tokenizer=self.get_bart_tokenizer()   ,						retrieval_workers=A   ,						)
									if init_retrieval:
											retriever.init_retrieval()
							return retriever




					def   A  (    self					:						int   ,						A					:						bool   ,						A					:						bool   ,						A					:						Optional[Any]=1_23_45       )  ->  Dict:
							lowercase_			:			Optional[Any]								=       self.get_dummy_dataset()
							lowercase_			:			Dict								=       RagConfig(
							    retrieval_vector_size=self.retrieval_vector_size   ,						question_encoder=DPRConfig().to_dict()   ,						generator=BartConfig().to_dict()   ,						index_name='''custom'''   ,						)
							if from_disk:
									lowercase_			:			List[Any]								=       os.path.join(self.tmpdirname   ,						'''dataset'''       )
									lowercase_			:			Optional[int]								=       os.path.join(self.tmpdirname   ,						'''index.faiss'''       )
									dataset.get_index('''embeddings'''       ).save(os.path.join(self.tmpdirname   ,						'''index.faiss'''       )       )
									dataset.drop_index('''embeddings'''       )
									dataset.save_to_disk(os.path.join(self.tmpdirname   ,						'''dataset'''       )       )
									del dataset
									lowercase_			:			Dict								=       RagPyTorchDistributedRetriever(
									    A   ,						question_encoder_tokenizer=self.get_dpr_tokenizer()   ,						generator_tokenizer=self.get_bart_tokenizer()   ,						)
							else:
									lowercase_			:			Optional[Any]								=       RagPyTorchDistributedRetriever(
									    A   ,						question_encoder_tokenizer=self.get_dpr_tokenizer()   ,						generator_tokenizer=self.get_bart_tokenizer()   ,						index=CustomHFIndex(config.retrieval_vector_size   ,						A       )   ,						)
							if init_retrieval:
									retriever.init_retrieval(A       )
							return retriever




					def   A  (    self					:						Any   ,						A					:						bool   ,						A					:						bool       )  ->  Optional[int]:
							# Have to run in local mode because sys.path modifications at top of
							# file are not propogated to remote workers.
							# https://stackoverflow.com/questions/54338013/parallel-import-a-python-file-from-sibling-folder
							ray.init(local_mode=A       )
							lowercase_			:			List[str]								=       self.get_dummy_dataset()
							lowercase_			:			Union[str, Any]								=       RagConfig(
							    retrieval_vector_size=self.retrieval_vector_size   ,						question_encoder=DPRConfig().to_dict()   ,						generator=BartConfig().to_dict()   ,						index_name='''custom'''   ,						)
							lowercase_			:			Union[str, Any]								=       ray.remote(A       )
							lowercase_			:			Union[str, Any]								=       [remote_cls.remote() for _ in range(1       )]
							if from_disk:
									lowercase_			:			Optional[Any]								=       os.path.join(self.tmpdirname   ,						'''dataset'''       )
									lowercase_			:			List[str]								=       os.path.join(self.tmpdirname   ,						'''index.faiss'''       )
									dataset.get_index('''embeddings'''       ).save(os.path.join(self.tmpdirname   ,						'''index.faiss'''       )       )
									dataset.drop_index('''embeddings'''       )
									dataset.save_to_disk(os.path.join(self.tmpdirname   ,						'''dataset'''       )       )
									del dataset
									lowercase_			:			Optional[Any]								=       RagRayDistributedRetriever(
									    A   ,						question_encoder_tokenizer=self.get_dpr_tokenizer()   ,						generator_tokenizer=self.get_bart_tokenizer()   ,						retrieval_workers=A   ,						index=CustomHFIndex.load_from_disk(
									        vector_size=config.retrieval_vector_size   ,						dataset_path=config.passages_path   ,						index_path=config.index_path   ,						)   ,						)
							else:
									lowercase_			:			Dict								=       RagRayDistributedRetriever(
									    A   ,						question_encoder_tokenizer=self.get_dpr_tokenizer()   ,						generator_tokenizer=self.get_bart_tokenizer()   ,						retrieval_workers=A   ,						index=CustomHFIndex(config.retrieval_vector_size   ,						A       )   ,						)
							if init_retrieval:
									retriever.init_retrieval()
							return retriever




					def   A  (    self					:						int   ,						A					:						RagRetriever   ,						A					:						np.array   ,						A					:						int       )  ->  None:
							lowercase_	, lowercase_	, lowercase_			:			Dict								=       retriever.retrieve(A   ,						n_docs=A       )
							self.assertEqual(retrieved_doc_embeds.shape   ,						(2, n_docs, self.retrieval_vector_size)       )
							self.assertEqual(len(A       )   ,						2       )
							self.assertEqual(sorted(doc_dicts[0]       )   ,						['''embeddings''', '''id''', '''text''', '''title''']       )
							self.assertEqual(len(doc_dicts[0]['''id''']       )   ,						A       )
							self.assertEqual(doc_dicts[0]['''id'''][0]   ,						'''1'''       )  # max inner product is reached with second doc
							self.assertEqual(doc_dicts[1]['''id'''][0]   ,						'''0'''       )  # max inner product is reached with first doc
							self.assertListEqual(doc_ids.tolist()   ,						[[1], [0]]       )




					def   A  (    self					:						int       )  ->  Optional[Any]:
							lowercase_			:			Dict								=       1
							lowercase_			:			List[str]								=       np.array(
							    [np.ones(self.retrieval_vector_size       ), -np.ones(self.retrieval_vector_size       )]   ,						dtype=np.floataa       )

							self.distributed_retriever_check(
							    self.get_dummy_pytorch_distributed_retriever(init_retrieval=A       )   ,						A   ,						A       )




					def   A  (    self					:						Optional[int]       )  ->  Tuple:
							lowercase_			:			str								=       1
							lowercase_			:			Tuple								=       np.array(
							    [np.ones(self.retrieval_vector_size       ), -np.ones(self.retrieval_vector_size       )]   ,						dtype=np.floataa       )

							self.distributed_retriever_check(
							    self.get_dummy_custom_hf_index_pytorch_retriever(init_retrieval=A   ,						from_disk=A       )   ,						A   ,						A   ,						)




					def   A  (    self					:						Dict       )  ->  int:
							lowercase_			:			Dict								=       1
							lowercase_			:			Optional[Any]								=       np.array(
							    [np.ones(self.retrieval_vector_size       ), -np.ones(self.retrieval_vector_size       )]   ,						dtype=np.floataa       )

							self.distributed_retriever_check(
							    self.get_dummy_custom_hf_index_pytorch_retriever(init_retrieval=A   ,						from_disk=A       )   ,						A   ,						A   ,						)




					@require_ray
					def   A  (    self					:						Optional[int]       )  ->  Dict:
							lowercase_			:			Optional[Any]								=       1
							lowercase_			:			Optional[Any]								=       np.array(
							    [np.ones(self.retrieval_vector_size       ), -np.ones(self.retrieval_vector_size       )]   ,						dtype=np.floataa       )

							self.distributed_retriever_check(
							    self.get_dummy_ray_distributed_retriever(init_retrieval=A       )   ,						A   ,						A       )
							ray.shutdown()




					@require_ray
					def   A  (    self					:						int       )  ->  str:
							lowercase_			:			str								=       1
							lowercase_			:			List[str]								=       np.array(
							    [np.ones(self.retrieval_vector_size       ), -np.ones(self.retrieval_vector_size       )]   ,						dtype=np.floataa       )
							with self.assertRaises(A       ):
									self.distributed_retriever_check(
									    self.get_dummy_custom_hf_index_ray_retriever(init_retrieval=A   ,						from_disk=A       )   ,						A   ,						A   ,						)
							ray.shutdown()






					@require_ray
					def   A  (    self					:						Any       )  ->  Dict:
							lowercase_			:			Optional[int]								=       1
							lowercase_			:			List[Any]								=       np.array(
							    [np.ones(self.retrieval_vector_size       ), -np.ones(self.retrieval_vector_size       )]   ,						dtype=np.floataa       )

							self.distributed_retriever_check(
							    self.get_dummy_custom_hf_index_ray_retriever(init_retrieval=A   ,						from_disk=A       )   ,						A   ,						A       )
							ray.shutdown()











"""simple docstring"""



from typing import Callable, List, Optional, Union

import torch

from ...models import UNetaDModel
from ...schedulers import CMStochasticIterativeScheduler
from ...utils import (
    is_accelerate_available,
    is_accelerate_version,
    logging,
    randn_tensor,
    replace_example_docstring,
)
from ..pipeline_utils import DiffusionPipeline, ImagePipelineOutput


__A		:	Union[str, Any]      =		logging.get_logger(__name__)  # pylint: disable=invalid-name


__A		:	str      =		'''
    Examples:
        ```py
        >>> import torch

        >>> from diffusers import ConsistencyModelPipeline

        >>> device = "cuda"
        >>> # Load the cd_imagenet64_l2 checkpoint.
        >>> model_id_or_path = "openai/diffusers-cd_imagenet64_l2"
        >>> pipe = ConsistencyModelPipeline.from_pretrained(model_id_or_path, torch_dtype=torch.float16)
        >>> pipe.to(device)

        >>> # Onestep Sampling
        >>> image = pipe(num_inference_steps=1).images[0]
        >>> image.save("cd_imagenet64_l2_onestep_sample.png")

        >>> # Onestep sampling, class-conditional image generation
        >>> # ImageNet-64 class label 145 corresponds to king penguins
        >>> image = pipe(num_inference_steps=1, class_labels=145).images[0]
        >>> image.save("cd_imagenet64_l2_onestep_sample_penguin.png")

        >>> # Multistep sampling, class-conditional image generation
        >>> # Timesteps can be explicitly specified; the particular timesteps below are from the original Github repo:
        >>> # https://github.com/openai/consistency_models/blob/main/scripts/launch.sh#L77
        >>> image = pipe(num_inference_steps=None, timesteps=[22, 0], class_labels=145).images[0]
        >>> image.save("cd_imagenet64_l2_multistep_sample_penguin.png")
        ```
'''





class  _UpperCAmelCase	(			_A							):




					def __init__(    self					:						Union[str, Any]   ,						A					:						UNetaDModel   ,						A					:						CMStochasticIterativeScheduler       )  ->  None:
							super().__init__()

							self.register_modules(
							    unet=A   ,						scheduler=A   ,						)

							lowercase_			:			Any								=       None




					def   A  (    self					:						Tuple   ,						A					:						Union[str, Any]=0       )  ->  Optional[Any]:
							if is_accelerate_available() and is_accelerate_version('''>='''   ,						'''0.14.0'''       ):
									from accelerate import cpu_offload
							else:
									raise ImportError('''`enable_sequential_cpu_offload` requires `accelerate v0.14.0` or higher'''       )

							lowercase_			:			Any								=       torch.device(F'''cuda:{gpu_id}'''       )

							if self.device.type != "cpu":
									self.to('''cpu'''   ,						silence_dtype_warnings=A       )
									torch.cuda.empty_cache()  # otherwise we don't see the memory savings (but they probably exist)

							for cpu_offloaded_model in [self.unet]:
									cpu_offload(A   ,						A       )

							if self.safety_checker is not None:
									cpu_offload(self.safety_checker   ,						execution_device=A   ,						offload_buffers=A       )




					def   A  (    self					:						List[Any]   ,						A					:						Optional[Any]=0       )  ->  List[Any]:
							if is_accelerate_available() and is_accelerate_version('''>='''   ,						'''0.17.0.dev0'''       ):
									from accelerate import cpu_offload_with_hook
							else:
									raise ImportError('''`enable_model_cpu_offload` requires `accelerate v0.17.0` or higher.'''       )

							lowercase_			:			List[str]								=       torch.device(F'''cuda:{gpu_id}'''       )

							if self.device.type != "cpu":
									self.to('''cpu'''   ,						silence_dtype_warnings=A       )
									torch.cuda.empty_cache()  # otherwise we don't see the memory savings (but they probably exist)

							lowercase_			:			Union[str, Any]								=       None
							for cpu_offloaded_model in [self.unet]:
									lowercase_	, lowercase_			:			Optional[int]								=       cpu_offload_with_hook(A   ,						A   ,						prev_module_hook=A       )

							if self.safety_checker is not None:
									lowercase_	, lowercase_			:			Any								=       cpu_offload_with_hook(self.safety_checker   ,						A   ,						prev_module_hook=A       )

							# We'll offload the last model manually.
							lowercase_			:			str								=       hook




					@property
					# Copied from diffusers.pipelines.stable_diffusion.pipeline_stable_diffusion.StableDiffusionPipeline._execution_device
					def   A  (    self					:						Any       )  ->  List[str]:
							if not hasattr(self.unet   ,						'''_hf_hook'''       ):
									return self.device
							for module in self.unet.modules():
									if (
									    hasattr(A   ,						'''_hf_hook'''       )
									    and hasattr(module._hf_hook   ,						'''execution_device'''       )
									    and module._hf_hook.execution_device is not None
									):
											return torch.device(module._hf_hook.execution_device       )
							return self.device




					def   A  (    self					:						List[str]   ,						A					:						int   ,						A					:						Union[str, Any]   ,						A					:						List[str]   ,						A					:						str   ,						A					:						int   ,						A					:						List[Any]   ,						A					:						Optional[Any]   ,						A					:						Optional[Any]=None       )  ->  Tuple:
							lowercase_			:			List[Any]								=       (batch_size, num_channels, height, width)
							if isinstance(A   ,						A       ) and len(A       ) != batch_size:
									raise ValueError(
									    F'''You have passed a list of generators of length {len(A       )}, but requested an effective batch'''
									    F''' size of {batch_size}. Make sure the batch size matches the length of the generators.'''       )

							if latents is None:
									lowercase_			:			Dict								=       randn_tensor(A   ,						generator=A   ,						device=A   ,						dtype=A       )
							else:
									lowercase_			:			Tuple								=       latents.to(device=A   ,						dtype=A       )

							# scale the initial noise by the standard deviation required by the scheduler
							lowercase_			:			List[str]								=       latents * self.scheduler.init_noise_sigma
							return latents




					def   A  (    self					:						Tuple   ,						A					:						torch.FloatTensor   ,						A					:						str = "pil"       )  ->  Optional[int]:
							if output_type not in ["pt", "np", "pil"]:
									raise ValueError(
									    F'''output_type={output_type} is not supported. Make sure to choose one of [\'pt\', \'np\', or \'pil\']'''       )

							# Equivalent to diffusers.VaeImageProcessor.denormalize
							lowercase_			:			Dict								=       (sample / 2 + 0.5).clamp(0   ,						1       )
							if output_type == "pt":
									return sample

							# Equivalent to diffusers.VaeImageProcessor.pt_to_numpy
							lowercase_			:			List[str]								=       sample.cpu().permute(0   ,						2   ,						3   ,						1       ).numpy()
							if output_type == "np":
									return sample

							# Output_type must be 'pil'
							lowercase_			:			List[Any]								=       self.numpy_to_pil(A       )
							return sample




					def   A  (    self					:						Optional[int]   ,						A					:						Any   ,						A					:						Union[str, Any]   ,						A					:						Optional[int]=None       )  ->  List[Any]:
							if self.unet.config.num_class_embeds is not None:
									if isinstance(A   ,						A       ):
											lowercase_			:			Optional[int]								=       torch.tensor(A   ,						dtype=torch.int       )
									elif isinstance(A   ,						A       ):
											assert batch_size == 1, "Batch size must be 1 if classes is an int"
											lowercase_			:			int								=       torch.tensor([class_labels]   ,						dtype=torch.int       )
									elif class_labels is None:
											# Randomly generate batch_size class labels
											# TODO: should use generator here? int analogue of randn_tensor is not exposed in ...utils
											lowercase_			:			Any								=       torch.randint(0   ,						self.unet.config.num_class_embeds   ,						size=(batch_size,)       )
									lowercase_			:			List[str]								=       class_labels.to(A       )
							else:
									lowercase_			:			Union[str, Any]								=       None
							return class_labels




					def   A  (    self					:						Dict   ,						A					:						Union[str, Any]   ,						A					:						Any   ,						A					:						Optional[int]   ,						A					:						Union[str, Any]   ,						A					:						Any   ,						A					:						int       )  ->  Union[str, Any]:
							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:
									logger.warning(
									    F'''Both `num_inference_steps`: {num_inference_steps} and `timesteps`: {timesteps} are supplied;'''
									    ''' `timesteps` will be used over `num_inference_steps`.'''       )

							if latents is not None:
									lowercase_			:			List[Any]								=       (batch_size, 3, img_size, img_size)
									if latents.shape != expected_shape:
											raise ValueError(F'''The shape of latents is {latents.shape} but is expected to be {expected_shape}.'''       )

							if (callback_steps is None) or (
							    callback_steps is not None and (not isinstance(A   ,						A       ) or callback_steps <= 0)
							):
									raise ValueError(
									    F'''`callback_steps` has to be a positive integer but is {callback_steps} of type'''
									    F''' {type(A       )}.'''       )






					@torch.no_grad()
					@replace_example_docstring(A       )
					def __call__(    self					:						Optional[Any]   ,						A					:						int = 1   ,						A					:						Optional[Union[torch.Tensor, List[int], int]] = None   ,						A					:						int = 1   ,						A					:						List[int] = None   ,						A					:						Optional[Union[torch.Generator, List[torch.Generator]]] = None   ,						A					:						Optional[torch.FloatTensor] = None   ,						A					:						Optional[str] = "pil"   ,						A					:						bool = True   ,						A					:						Optional[Callable[[int, int, torch.FloatTensor], None]] = None   ,						A					:						int = 1   ,						)  ->  Tuple:
							lowercase_			:			Dict								=       self.unet.config.sample_size
							lowercase_			:			Union[str, Any]								=       self._execution_device

							# 1. Check inputs
							self.check_inputs(A   ,						A   ,						A   ,						A   ,						A   ,						A       )

							# 2. Prepare image latents
							# Sample image latents x_0 ~ N(0, sigma_0^2 * I)
							lowercase_			:			Optional[int]								=       self.prepare_latents(
							    batch_size=A   ,						num_channels=self.unet.config.in_channels   ,						height=A   ,						width=A   ,						dtype=self.unet.dtype   ,						device=A   ,						generator=A   ,						latents=A   ,						)

							# 3. Handle class_labels for class-conditional models
							lowercase_			:			List[str]								=       self.prepare_class_labels(A   ,						A   ,						class_labels=A       )

							# 4. Prepare timesteps
							if timesteps is not None:
									self.scheduler.set_timesteps(timesteps=A   ,						device=A       )
									lowercase_			:			Optional[int]								=       self.scheduler.timesteps
									lowercase_			:			List[Any]								=       len(A       )
							else:
									self.scheduler.set_timesteps(A       )
									lowercase_			:			Any								=       self.scheduler.timesteps

							# 5. Denoising loop
							# Multistep sampling: implements Algorithm 1 in the paper
							with self.progress_bar(total=A       ) as progress_bar:
									for i, t in enumerate(A       ):
											lowercase_			:			List[str]								=       self.scheduler.scale_model_input(A   ,						A       )
											lowercase_			:			Optional[Any]								=       self.unet(A   ,						A   ,						class_labels=A   ,						return_dict=A       )[0]

											lowercase_			:			Tuple								=       self.scheduler.step(A   ,						A   ,						A   ,						generator=A       )[0]

											# call the callback, if provided
											progress_bar.update()
											if callback is not None and i % callback_steps == 0:
													callback(A   ,						A   ,						A       )

        # 6. Post-process image sample
							lowercase_			:			Union[str, Any]								=       self.postprocess_image(A   ,						output_type=A       )

							# Offload last model to CPU
							if hasattr(self   ,						'''final_offload_hook'''       ) and self.final_offload_hook is not None:
									self.final_offload_hook.offload()

							if not return_dict:
									return (image,)

							return ImagePipelineOutput(images=A       )



