


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

import numpy as np
from datasets import Dataset

from transformers import is_faiss_available
from transformers.models.bart.configuration_bart import BartConfig
from transformers.models.bart.tokenization_bart import BartTokenizer
from transformers.models.bert.tokenization_bert import VOCAB_FILES_NAMES as DPR_VOCAB_FILES_NAMES
from transformers.models.dpr.configuration_dpr import DPRConfig
from transformers.models.dpr.tokenization_dpr import DPRContextEncoderTokenizer, DPRQuestionEncoderTokenizer
from transformers.models.rag.configuration_rag import RagConfig
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_faiss, require_sentencepiece, require_tokenizers, require_torch


if is_faiss_available():
  import faiss







@require_faiss
class   __lowerCAmelCase			(					__UpperCAmelCase  ):





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





      a__	:						Optional[int]      		=  tempfile.mkdtemp()
      a__	:						Tuple      		=  8

      # DPR tok
      a__	:						int      		=  [
          "[UNK]",
          "[CLS]",
          "[SEP]",
          "[PAD]",
          "[MASK]",
          "want",
          "##want",
          "##ed",
          "wa",
          "un",
          "runn",
          "##ing",
          ",",
          "low",
          "lowest",
      ]
      a__	:						Tuple      		=  os.path.join(self.tmpdirname					,			"dpr_tokenizer"      )
      os.makedirs(UpperCAmelCase_					,			exist_ok=UpperCAmelCase_      )
      a__	:						Dict      		=  os.path.join(UpperCAmelCase_					,			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
      a__	:						List[Any]      		=  [
          "l",
          "o",
          "w",
          "e",
          "r",
          "s",
          "t",
          "i",
          "d",
          "n",
          "\u0120",
          "\u0120l",
          "\u0120n",
          "\u0120lo",
          "\u0120low",
          "er",
          "\u0120lowest",
          "\u0120newer",
          "\u0120wider",
          "<unk>",
      ]
      a__	:						Union[str, Any]      		=  dict(zip(UpperCAmelCase_					,			range(len(UpperCAmelCase_      )      )      )      )
      a__	:						List[str]      		=  ["#version: 0.2", "\u0120 l", "\u0120l o", "\u0120lo w", "e r", ""]
      a__	:						int      		=  {"unk_token": "<unk>"}

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





   def 	_snake_case ( self      )				->  int:
      """simple docstring"""





      return DPRQuestionEncoderTokenizer.from_pretrained(os.path.join(self.tmpdirname					,			"dpr_tokenizer"      )      )





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





      return DPRContextEncoderTokenizer.from_pretrained(os.path.join(self.tmpdirname					,			"dpr_tokenizer"      )      )





   def 	_snake_case ( self      )				->  Any:
      """simple docstring"""





      return BartTokenizer.from_pretrained(os.path.join(self.tmpdirname					,			"bart_tokenizer"      )      )





   def 	_snake_case ( self      )				->  Any:
      """simple docstring"""





      shutil.rmtree(self.tmpdirname      )





   def 	_snake_case ( self      )				->  int:
      """simple docstring"""





      a__	:						Dict      		=  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 	_snake_case ( self      )				->  Optional[Any]:
      """simple docstring"""





      a__	:						str      		=  self.get_dummy_dataset()
      a__	:						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:
         a__	:						Optional[int]      		=  dataset
         a__	:						Dict      		=  RagRetriever(
             UpperCAmelCase_					,			question_encoder_tokenizer=self.get_dpr_tokenizer()					,			generator_tokenizer=self.get_bart_tokenizer()					,			)
      return retriever





   def 	_snake_case ( self					,			snake_case      )				->  Optional[int]:
      """simple docstring"""





      a__	:						List[str]      		=  self.get_dummy_dataset()
      a__	:						str      		=  RagConfig(
          retrieval_vector_size=self.retrieval_vector_size					,			question_encoder=DPRConfig().to_dict()					,			generator=BartConfig().to_dict()					,			index_name="custom"					,			)
      if from_disk:
         a__	:						Tuple      		=  os.path.join(self.tmpdirname					,			"dataset"      )
         a__	:						Dict      		=  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
         a__	:						str      		=  RagRetriever(
             UpperCAmelCase_					,			question_encoder_tokenizer=self.get_dpr_tokenizer()					,			generator_tokenizer=self.get_bart_tokenizer()					,			)
      else:
         a__	:						List[Any]      		=  RagRetriever(
             UpperCAmelCase_					,			question_encoder_tokenizer=self.get_dpr_tokenizer()					,			generator_tokenizer=self.get_bart_tokenizer()					,			index=CustomHFIndex(config.retrieval_vector_size					,			UpperCAmelCase_      )					,			)
      return retriever





   def 	_snake_case ( self      )				->  Dict:
      """simple docstring"""





      a__	:						List[Any]      		=  Dataset.from_dict(
          {
              "id": ["0", "1"],
              "text": ["foo", "bar"],
              "title": ["Foo", "Bar"],
              "embeddings": [np.ones(self.retrieval_vector_size + 1      ), 2 * np.ones(self.retrieval_vector_size + 1      )],
          }      )
      dataset.add_faiss_index("embeddings"					,			string_factory="Flat"					,			metric_type=faiss.METRIC_INNER_PRODUCT      )

      a__	:						Tuple      		=  os.path.join(self.tmpdirname					,			"hf_bert_base.hnswSQ8_correct_phi_128.c_index"      )
      dataset.save_faiss_index("embeddings"					,			index_file_name + ".index.dpr"      )
      pickle.dump(dataset["id"]					,			open(index_file_name + ".index_meta.dpr"					,			"wb"      )      )

      a__	:						int      		=  os.path.join(self.tmpdirname					,			"psgs_w100.tsv.pkl"      )
      a__	:						Dict      		=  {sample["id"]: [sample["text"], sample["title"]] for sample in dataset}
      pickle.dump(UpperCAmelCase_					,			open(UpperCAmelCase_					,			"wb"      )      )

      a__	:						List[str]      		=  RagConfig(
          retrieval_vector_size=self.retrieval_vector_size					,			question_encoder=DPRConfig().to_dict()					,			generator=BartConfig().to_dict()					,			index_name="legacy"					,			index_path=self.tmpdirname					,			)
      a__	:						Optional[int]      		=  RagRetriever(
          UpperCAmelCase_					,			question_encoder_tokenizer=self.get_dpr_tokenizer()					,			generator_tokenizer=self.get_bart_tokenizer()      )
      return retriever





   def 	_snake_case ( self      )				->  Any:
      """simple docstring"""





      a__	:						str      		=  1
      a__	:						List[Any]      		=  self.get_dummy_canonical_hf_index_retriever()
      a__	:						Optional[int]      		=  np.array(
          [np.ones(self.retrieval_vector_size      ), -np.ones(self.retrieval_vector_size      )]					,			dtype=np.floataa      )
      a__		,     a__		,     a__	:						Dict      		=  retriever.retrieve(UpperCAmelCase_					,			n_docs=UpperCAmelCase_      )
      self.assertEqual(retrieved_doc_embeds.shape					,			(2, n_docs, self.retrieval_vector_size)      )
      self.assertEqual(len(UpperCAmelCase_      )					,			2      )
      self.assertEqual(sorted(doc_dicts[0]      )					,			["embeddings", "id", "text", "title"]      )
      self.assertEqual(len(doc_dicts[0]["id"]      )					,			UpperCAmelCase_      )
      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 	_snake_case ( self      )				->  str:
      """simple docstring"""





      a__	:						Any      		=  self.get_dummy_canonical_hf_index_retriever()
      with tempfile.TemporaryDirectory() as tmp_dirname:
         with patch("transformers.models.rag.retrieval_rag.load_dataset"      ) as mock_load_dataset:
            a__	:						Optional[Any]      		=  self.get_dummy_dataset()
            retriever.save_pretrained(UpperCAmelCase_      )
            a__	:						str      		=  RagRetriever.from_pretrained(UpperCAmelCase_      )
            self.assertIsInstance(UpperCAmelCase_					,			UpperCAmelCase_      )
            a__	:						Any      		=  np.array(
                [np.ones(self.retrieval_vector_size      ), -np.ones(self.retrieval_vector_size      )]					,			dtype=np.floataa      )
            a__	:						Optional[Any]      		=  retriever.retrieve(UpperCAmelCase_					,			n_docs=1      )
            self.assertTrue(out is not None      )





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





      a__	:						List[Any]      		=  1
      a__	:						Union[str, Any]      		=  self.get_dummy_custom_hf_index_retriever(from_disk=UpperCAmelCase_      )
      a__	:						Union[str, Any]      		=  np.array(
          [np.ones(self.retrieval_vector_size      ), -np.ones(self.retrieval_vector_size      )]					,			dtype=np.floataa      )
      a__		,     a__		,     a__	:						Optional[Any]      		=  retriever.retrieve(UpperCAmelCase_					,			n_docs=UpperCAmelCase_      )
      self.assertEqual(retrieved_doc_embeds.shape					,			(2, n_docs, self.retrieval_vector_size)      )
      self.assertEqual(len(UpperCAmelCase_      )					,			2      )
      self.assertEqual(sorted(doc_dicts[0]      )					,			["embeddings", "id", "text", "title"]      )
      self.assertEqual(len(doc_dicts[0]["id"]      )					,			UpperCAmelCase_      )
      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 	_snake_case ( self      )				->  Tuple:
      """simple docstring"""





      a__	:						List[str]      		=  self.get_dummy_custom_hf_index_retriever(from_disk=UpperCAmelCase_      )
      with tempfile.TemporaryDirectory() as tmp_dirname:
         retriever.save_pretrained(UpperCAmelCase_      )
         a__	:						List[str]      		=  RagRetriever.from_pretrained(UpperCAmelCase_      )
         self.assertIsInstance(UpperCAmelCase_					,			UpperCAmelCase_      )
         a__	:						List[str]      		=  np.array(
             [np.ones(self.retrieval_vector_size      ), -np.ones(self.retrieval_vector_size      )]					,			dtype=np.floataa      )
         a__	:						Any      		=  retriever.retrieve(UpperCAmelCase_					,			n_docs=1      )
         self.assertTrue(out is not None      )





   def 	_snake_case ( self      )				->  str:
      """simple docstring"""





      a__	:						List[str]      		=  1
      a__	:						Optional[Any]      		=  self.get_dummy_custom_hf_index_retriever(from_disk=UpperCAmelCase_      )
      a__	:						Optional[int]      		=  np.array(
          [np.ones(self.retrieval_vector_size      ), -np.ones(self.retrieval_vector_size      )]					,			dtype=np.floataa      )
      a__		,     a__		,     a__	:						Union[str, Any]      		=  retriever.retrieve(UpperCAmelCase_					,			n_docs=UpperCAmelCase_      )
      self.assertEqual(retrieved_doc_embeds.shape					,			(2, n_docs, self.retrieval_vector_size)      )
      self.assertEqual(len(UpperCAmelCase_      )					,			2      )
      self.assertEqual(sorted(doc_dicts[0]      )					,			["embeddings", "id", "text", "title"]      )
      self.assertEqual(len(doc_dicts[0]["id"]      )					,			UpperCAmelCase_      )
      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 	_snake_case ( self      )				->  Optional[int]:
      """simple docstring"""





      a__	:						str      		=  self.get_dummy_custom_hf_index_retriever(from_disk=UpperCAmelCase_      )
      with tempfile.TemporaryDirectory() as tmp_dirname:
         retriever.save_pretrained(UpperCAmelCase_      )
         a__	:						Optional[int]      		=  RagRetriever.from_pretrained(UpperCAmelCase_      )
         self.assertIsInstance(UpperCAmelCase_					,			UpperCAmelCase_      )
         a__	:						Union[str, Any]      		=  np.array(
             [np.ones(self.retrieval_vector_size      ), -np.ones(self.retrieval_vector_size      )]					,			dtype=np.floataa      )
         a__	:						Tuple      		=  retriever.retrieve(UpperCAmelCase_					,			n_docs=1      )
         self.assertTrue(out is not None      )





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





      a__	:						Any      		=  1
      a__	:						Optional[int]      		=  self.get_dummy_legacy_index_retriever()
      a__	:						int      		=  np.array(
          [np.ones(self.retrieval_vector_size      ), -np.ones(self.retrieval_vector_size      )]					,			dtype=np.floataa      )
      a__		,     a__		,     a__	:						Tuple      		=  retriever.retrieve(UpperCAmelCase_					,			n_docs=UpperCAmelCase_      )
      self.assertEqual(retrieved_doc_embeds.shape					,			(2, n_docs, self.retrieval_vector_size)      )
      self.assertEqual(len(UpperCAmelCase_      )					,			2      )
      self.assertEqual(sorted(doc_dicts[0]      )					,			["text", "title"]      )
      self.assertEqual(len(doc_dicts[0]["text"]      )					,			UpperCAmelCase_      )
      self.assertEqual(doc_dicts[0]["text"][0]					,			"bar"      )  # max inner product is reached with second doc
      self.assertEqual(doc_dicts[1]["text"][0]					,			"foo"      )  # max inner product is reached with first doc
      self.assertListEqual(doc_ids.tolist()					,			[[1], [0]]      )





   def 	_snake_case ( self      )				->  Optional[Any]:
      """simple docstring"""





      a__	:						Tuple      		=  self.get_dummy_legacy_index_retriever()
      with tempfile.TemporaryDirectory() as tmp_dirname:
         retriever.save_pretrained(UpperCAmelCase_      )
         a__	:						Union[str, Any]      		=  RagRetriever.from_pretrained(UpperCAmelCase_      )
         self.assertIsInstance(UpperCAmelCase_					,			UpperCAmelCase_      )
         a__	:						Union[str, Any]      		=  np.array(
             [np.ones(self.retrieval_vector_size      ), -np.ones(self.retrieval_vector_size      )]					,			dtype=np.floataa      )
         a__	:						List[str]      		=  retriever.retrieve(UpperCAmelCase_					,			n_docs=1      )
         self.assertTrue(out is not None      )





   @require_torch
   @require_tokenizers
   @require_sentencepiece
   def 	_snake_case ( self      )				->  int:
      """simple docstring"""





      import torch

      a__	:						Any      		=  1
      a__	:						Dict      		=  self.get_dummy_canonical_hf_index_retriever()
      a__	:						Optional[int]      		=  [[5, 7], [10, 11]]
      a__	:						Tuple      		=  np.array(
          [np.ones(self.retrieval_vector_size      ), -np.ones(self.retrieval_vector_size      )]					,			dtype=np.floataa      )
      a__	:						str      		=  retriever(UpperCAmelCase_					,			UpperCAmelCase_					,			prefix=retriever.config.generator.prefix					,			n_docs=UpperCAmelCase_      )
      a__		,     a__		,     a__	:						List[str]      		=  (
          out["context_input_ids"],
          out["context_attention_mask"],
          out["retrieved_doc_embeds"],
      )
      self.assertEqual(retrieved_doc_embeds.shape					,			(2, n_docs, self.retrieval_vector_size)      )
      self.assertIsInstance(UpperCAmelCase_					,			UpperCAmelCase_      )
      self.assertIsInstance(UpperCAmelCase_					,			UpperCAmelCase_      )
      self.assertIsInstance(UpperCAmelCase_					,			np.ndarray      )

      a__	:						str      		=  retriever(
          UpperCAmelCase_					,			UpperCAmelCase_					,			prefix=retriever.config.generator.prefix					,			n_docs=UpperCAmelCase_					,			return_tensors="pt"					,			)
      a__		,     a__		,     a__		,     a__	:						Union[str, Any]      		=  (  # noqa: F841
          out["context_input_ids"],
          out["context_attention_mask"],
          out["retrieved_doc_embeds"],
          out["doc_ids"],
      )
      self.assertEqual(retrieved_doc_embeds.shape					,			(2, n_docs, self.retrieval_vector_size)      )
      self.assertIsInstance(UpperCAmelCase_					,			torch.Tensor      )
      self.assertIsInstance(UpperCAmelCase_					,			torch.Tensor      )
      self.assertIsInstance(UpperCAmelCase_					,			torch.Tensor      )





   @require_torch
   @require_tokenizers
   @require_sentencepiece
   def 	_snake_case ( self      )				->  Optional[int]:
      """simple docstring"""





      a__	:						Optional[Any]      		=  self.get_dpr_ctx_encoder_tokenizer()
      a__	:						List[str]      		=  1
      a__	:						Any      		=  self.get_dummy_custom_hf_index_retriever(from_disk=UpperCAmelCase_      )
      retriever.set_ctx_encoder_tokenizer(UpperCAmelCase_      )

      a__	:						Optional[Any]      		=  [[5, 7], [10, 11]]
      a__	:						Optional[Any]      		=  np.array(
          [np.ones(self.retrieval_vector_size      ), -np.ones(self.retrieval_vector_size      )]					,			dtype=np.floataa      )
      a__	:						Tuple      		=  retriever(UpperCAmelCase_					,			UpperCAmelCase_					,			prefix=retriever.config.generator.prefix					,			n_docs=UpperCAmelCase_      )

      self.assertEqual(
          len(UpperCAmelCase_      )					,			6      )  # check whether the retriever output consist of 6 attributes including tokenized docs
      self.assertEqual(
          all(k in out for k in ("tokenized_doc_ids", "tokenized_doc_attention_mask")      )					,			UpperCAmelCase_      )  # check for doc token related keys in dictionary.








from typing import Any, Callable, Dict, List, Optional, Union

import torch
from transformers import CLIPImageProcessor, CLIPTextModel, CLIPTokenizer

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    DiffusionPipeline,
    LMSDiscreteScheduler,
    PNDMScheduler,
    StableDiffusionPipeline,
    UNetaDConditionModel,
)
from diffusers.pipelines.stable_diffusion import StableDiffusionPipelineOutput
from diffusers.pipelines.stable_diffusion.safety_checker import StableDiffusionSafetyChecker


SCREAMING_SNAKE_CASE__       :			Dict			      =						"""CompVis/stable-diffusion-v1-1"""
SCREAMING_SNAKE_CASE__       :			Dict			      =						"""CompVis/stable-diffusion-v1-2"""
SCREAMING_SNAKE_CASE__       :			Tuple			      =						"""CompVis/stable-diffusion-v1-3"""
SCREAMING_SNAKE_CASE__       :			str			      =						"""CompVis/stable-diffusion-v1-4"""







class   __lowerCAmelCase			(					_UpperCamelCase  ):





   def __init__( self					,			snake_case					,			snake_case					,			snake_case					,			snake_case					,			snake_case					,			snake_case					,			snake_case					,			snake_case = True					,			)				->  Any:
      """simple docstring"""





      super()._init_()

      a__	:						Union[str, Any]      		=  StableDiffusionPipeline.from_pretrained(snake_case      )
      a__	:						Optional[Any]      		=  StableDiffusionPipeline.from_pretrained(snake_case      )
      a__	:						Optional[Any]      		=  StableDiffusionPipeline.from_pretrained(snake_case      )
      a__	:						int      		=  StableDiffusionPipeline(
          vae=snake_case					,			text_encoder=snake_case					,			tokenizer=snake_case					,			unet=snake_case					,			scheduler=snake_case					,			safety_checker=snake_case					,			feature_extractor=snake_case					,			requires_safety_checker=snake_case					,			)

      self.register_modules(pipelinea=self.pipea					,			pipelinea=self.pipea					,			pipelinea=self.pipea					,			pipelinea=self.pipea      )





   @property
   def 	_snake_case ( self      )				->  Dict[str, Any]:
      """simple docstring"""





      return {k: getattr(self					,			snake_case      ) for k in self.config.keys() if not k.startswith("_"      )}





   def 	_snake_case ( self					,			snake_case = "auto"      )				->  Optional[Any]:
      """simple docstring"""





      if slice_size == "auto":
         # half the attention head size is usually a good trade-off between
         # speed and memory
         a__	:						List[Any]      		=  self.unet.config.attention_head_dim // 2
      self.unet.set_attention_slice(snake_case      )





   def 	_snake_case ( self      )				->  Tuple:
      """simple docstring"""





      self.enable_attention_slicing(snake_case      )





   @torch.no_grad()
   def 	_snake_case ( self					,			snake_case					,			snake_case = 512					,			snake_case = 512					,			snake_case = 50					,			snake_case = 7.5					,			snake_case = None					,			snake_case = 1					,			snake_case = 0.0					,			snake_case = None					,			snake_case = None					,			snake_case = "pil"					,			snake_case = True					,			snake_case = None					,			snake_case = 1					,			**snake_case					,			)				->  Optional[int]:
      """simple docstring"""





      return self.pipea(
          prompt=snake_case					,			height=snake_case					,			width=snake_case					,			num_inference_steps=snake_case					,			guidance_scale=snake_case					,			negative_prompt=snake_case					,			num_images_per_prompt=snake_case					,			eta=snake_case					,			generator=snake_case					,			latents=snake_case					,			output_type=snake_case					,			return_dict=snake_case					,			callback=snake_case					,			callback_steps=snake_case					,			**snake_case					,			)





   @torch.no_grad()
   def 	_snake_case ( self					,			snake_case					,			snake_case = 512					,			snake_case = 512					,			snake_case = 50					,			snake_case = 7.5					,			snake_case = None					,			snake_case = 1					,			snake_case = 0.0					,			snake_case = None					,			snake_case = None					,			snake_case = "pil"					,			snake_case = True					,			snake_case = None					,			snake_case = 1					,			**snake_case					,			)				->  Dict:
      """simple docstring"""





      return self.pipea(
          prompt=snake_case					,			height=snake_case					,			width=snake_case					,			num_inference_steps=snake_case					,			guidance_scale=snake_case					,			negative_prompt=snake_case					,			num_images_per_prompt=snake_case					,			eta=snake_case					,			generator=snake_case					,			latents=snake_case					,			output_type=snake_case					,			return_dict=snake_case					,			callback=snake_case					,			callback_steps=snake_case					,			**snake_case					,			)





   @torch.no_grad()
   def 	_snake_case ( self					,			snake_case					,			snake_case = 512					,			snake_case = 512					,			snake_case = 50					,			snake_case = 7.5					,			snake_case = None					,			snake_case = 1					,			snake_case = 0.0					,			snake_case = None					,			snake_case = None					,			snake_case = "pil"					,			snake_case = True					,			snake_case = None					,			snake_case = 1					,			**snake_case					,			)				->  Union[str, Any]:
      """simple docstring"""





      return self.pipea(
          prompt=snake_case					,			height=snake_case					,			width=snake_case					,			num_inference_steps=snake_case					,			guidance_scale=snake_case					,			negative_prompt=snake_case					,			num_images_per_prompt=snake_case					,			eta=snake_case					,			generator=snake_case					,			latents=snake_case					,			output_type=snake_case					,			return_dict=snake_case					,			callback=snake_case					,			callback_steps=snake_case					,			**snake_case					,			)





   @torch.no_grad()
   def 	_snake_case ( self					,			snake_case					,			snake_case = 512					,			snake_case = 512					,			snake_case = 50					,			snake_case = 7.5					,			snake_case = None					,			snake_case = 1					,			snake_case = 0.0					,			snake_case = None					,			snake_case = None					,			snake_case = "pil"					,			snake_case = True					,			snake_case = None					,			snake_case = 1					,			**snake_case					,			)				->  Dict:
      """simple docstring"""





      return self.pipea(
          prompt=snake_case					,			height=snake_case					,			width=snake_case					,			num_inference_steps=snake_case					,			guidance_scale=snake_case					,			negative_prompt=snake_case					,			num_images_per_prompt=snake_case					,			eta=snake_case					,			generator=snake_case					,			latents=snake_case					,			output_type=snake_case					,			return_dict=snake_case					,			callback=snake_case					,			callback_steps=snake_case					,			**snake_case					,			)





   @torch.no_grad()
   def 	_snake_case ( self					,			snake_case					,			snake_case = 512					,			snake_case = 512					,			snake_case = 50					,			snake_case = 7.5					,			snake_case = None					,			snake_case = 1					,			snake_case = 0.0					,			snake_case = None					,			snake_case = None					,			snake_case = "pil"					,			snake_case = True					,			snake_case = None					,			snake_case = 1					,			**snake_case					,			)				->  Dict:
      """simple docstring"""





      a__	:						Any      		=  "cuda" if torch.cuda.is_available() else "cpu"
      self.to(snake_case      )

      # Checks if the height and width are divisible by 8 or not
      if height % 8 != 0 or width % 8 != 0:
         raise ValueError(F"""`height` and `width` must be divisible by 8 but are {height} and {width}."""      )

      # Get first result from Stable Diffusion Checkpoint v1.1
      a__	:						Any      		=  self.textaimg_sda_a(
          prompt=snake_case					,			height=snake_case					,			width=snake_case					,			num_inference_steps=snake_case					,			guidance_scale=snake_case					,			negative_prompt=snake_case					,			num_images_per_prompt=snake_case					,			eta=snake_case					,			generator=snake_case					,			latents=snake_case					,			output_type=snake_case					,			return_dict=snake_case					,			callback=snake_case					,			callback_steps=snake_case					,			**snake_case					,			)

      # Get first result from Stable Diffusion Checkpoint v1.2
      a__	:						List[Any]      		=  self.textaimg_sda_a(
          prompt=snake_case					,			height=snake_case					,			width=snake_case					,			num_inference_steps=snake_case					,			guidance_scale=snake_case					,			negative_prompt=snake_case					,			num_images_per_prompt=snake_case					,			eta=snake_case					,			generator=snake_case					,			latents=snake_case					,			output_type=snake_case					,			return_dict=snake_case					,			callback=snake_case					,			callback_steps=snake_case					,			**snake_case					,			)

      # Get first result from Stable Diffusion Checkpoint v1.3
      a__	:						Optional[Any]      		=  self.textaimg_sda_a(
          prompt=snake_case					,			height=snake_case					,			width=snake_case					,			num_inference_steps=snake_case					,			guidance_scale=snake_case					,			negative_prompt=snake_case					,			num_images_per_prompt=snake_case					,			eta=snake_case					,			generator=snake_case					,			latents=snake_case					,			output_type=snake_case					,			return_dict=snake_case					,			callback=snake_case					,			callback_steps=snake_case					,			**snake_case					,			)

      # Get first result from Stable Diffusion Checkpoint v1.4
      a__	:						Dict      		=  self.textaimg_sda_a(
          prompt=snake_case					,			height=snake_case					,			width=snake_case					,			num_inference_steps=snake_case					,			guidance_scale=snake_case					,			negative_prompt=snake_case					,			num_images_per_prompt=snake_case					,			eta=snake_case					,			generator=snake_case					,			latents=snake_case					,			output_type=snake_case					,			return_dict=snake_case					,			callback=snake_case					,			callback_steps=snake_case					,			**snake_case					,			)

      # Get all result images into a single list and pass it via StableDiffusionPipelineOutput for final result
      return StableDiffusionPipelineOutput([resa[0], resa[0], resa[0], resa[0]]      )

