


import collections
from typing import List, Optional, Union

from ...tokenization_utils_base import BatchEncoding
from ...utils import TensorType, add_end_docstrings, add_start_docstrings, logging
from ..bert.tokenization_bert_fast import BertTokenizerFast
from .tokenization_dpr import DPRContextEncoderTokenizer, DPRQuestionEncoderTokenizer, DPRReaderTokenizer


_snake_case					:		Tuple      =    logging.get_logger(__name__)

_snake_case					:		Optional[Any]      =    {"vocab_file": "vocab.txt", "tokenizer_file": "tokenizer.json"}

_snake_case					:		List[Any]      =    {
    "vocab_file": {
        "facebook/dpr-ctx_encoder-single-nq-base": (
            "https://huggingface.co/facebook/dpr-ctx_encoder-single-nq-base/resolve/main/vocab.txt"
        ),
        "facebook/dpr-ctx_encoder-multiset-base": (
            "https://huggingface.co/facebook/dpr-ctx_encoder-multiset-base/resolve/main/vocab.txt"
        ),
    },
    "tokenizer_file": {
        "facebook/dpr-ctx_encoder-single-nq-base": (
            "https://huggingface.co/facebook/dpr-ctx_encoder-single-nq-base/resolve/main/tokenizer.json"
        ),
        "facebook/dpr-ctx_encoder-multiset-base": (
            "https://huggingface.co/facebook/dpr-ctx_encoder-multiset-base/resolve/main/tokenizer.json"
        ),
    },
}
_snake_case					:		Any      =    {
    "vocab_file": {
        "facebook/dpr-question_encoder-single-nq-base": (
            "https://huggingface.co/facebook/dpr-question_encoder-single-nq-base/resolve/main/vocab.txt"
        ),
        "facebook/dpr-question_encoder-multiset-base": (
            "https://huggingface.co/facebook/dpr-question_encoder-multiset-base/resolve/main/vocab.txt"
        ),
    },
    "tokenizer_file": {
        "facebook/dpr-question_encoder-single-nq-base": (
            "https://huggingface.co/facebook/dpr-question_encoder-single-nq-base/resolve/main/tokenizer.json"
        ),
        "facebook/dpr-question_encoder-multiset-base": (
            "https://huggingface.co/facebook/dpr-question_encoder-multiset-base/resolve/main/tokenizer.json"
        ),
    },
}
_snake_case					:		List[Any]      =    {
    "vocab_file": {
        "facebook/dpr-reader-single-nq-base": (
            "https://huggingface.co/facebook/dpr-reader-single-nq-base/resolve/main/vocab.txt"
        ),
        "facebook/dpr-reader-multiset-base": (
            "https://huggingface.co/facebook/dpr-reader-multiset-base/resolve/main/vocab.txt"
        ),
    },
    "tokenizer_file": {
        "facebook/dpr-reader-single-nq-base": (
            "https://huggingface.co/facebook/dpr-reader-single-nq-base/resolve/main/tokenizer.json"
        ),
        "facebook/dpr-reader-multiset-base": (
            "https://huggingface.co/facebook/dpr-reader-multiset-base/resolve/main/tokenizer.json"
        ),
    },
}

_snake_case					:		str      =    {
    "facebook/dpr-ctx_encoder-single-nq-base": 512,
    "facebook/dpr-ctx_encoder-multiset-base": 512,
}
_snake_case					:		Optional[int]      =    {
    "facebook/dpr-question_encoder-single-nq-base": 512,
    "facebook/dpr-question_encoder-multiset-base": 512,
}
_snake_case					:		Optional[int]      =    {
    "facebook/dpr-reader-single-nq-base": 512,
    "facebook/dpr-reader-multiset-base": 512,
}


_snake_case					:		str      =    {
    "facebook/dpr-ctx_encoder-single-nq-base": {"do_lower_case": True},
    "facebook/dpr-ctx_encoder-multiset-base": {"do_lower_case": True},
}
_snake_case					:		List[Any]      =    {
    "facebook/dpr-question_encoder-single-nq-base": {"do_lower_case": True},
    "facebook/dpr-question_encoder-multiset-base": {"do_lower_case": True},
}
_snake_case					:		int      =    {
    "facebook/dpr-reader-single-nq-base": {"do_lower_case": True},
    "facebook/dpr-reader-multiset-base": {"do_lower_case": True},
}

class     a		(_lowerCAmelCase				):




							"""simple docstring"""


							__UpperCAmelCase   :							str					  =  VOCAB_FILES_NAMES
							__UpperCAmelCase   :							Dict					  =  CONTEXT_ENCODER_PRETRAINED_VOCAB_FILES_MAP
							__UpperCAmelCase   :							List[str]					  =  CONTEXT_ENCODER_PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
							__UpperCAmelCase   :							Union[str, Any]					  =  CONTEXT_ENCODER_PRETRAINED_INIT_CONFIGURATION
							__UpperCAmelCase   :							Optional[Any]					  =  DPRContextEncoderTokenizer

class     a		(_lowerCAmelCase				):




							"""simple docstring"""


							__UpperCAmelCase   :							str					  =  VOCAB_FILES_NAMES
							__UpperCAmelCase   :							int					  =  QUESTION_ENCODER_PRETRAINED_VOCAB_FILES_MAP
							__UpperCAmelCase   :							Optional[int]					  =  QUESTION_ENCODER_PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
							__UpperCAmelCase   :							Any					  =  QUESTION_ENCODER_PRETRAINED_INIT_CONFIGURATION
							__UpperCAmelCase   :							Dict					  =  DPRQuestionEncoderTokenizer
_snake_case					:		List[str]      =    collections.namedtuple(
    "DPRSpanPrediction", ["span_score", "relevance_score", "doc_id", "start_index", "end_index", "text"]
)

_snake_case					:		List[Any]      =    collections.namedtuple("DPRReaderOutput", ["start_logits", "end_logits", "relevance_logits"])


_snake_case					:		List[Any]      =    R"\n    Return a dictionary with the token ids of the input strings and other information to give to `.decode_best_spans`.\n    It converts the strings of a question and different passages (title and text) in a sequence of IDs (integers),\n    using the tokenizer and vocabulary. The resulting `input_ids` is a matrix of size `(n_passages, sequence_length)`\n    with the format:\n\n    [CLS] <question token ids> [SEP] <titles ids> [SEP] <texts ids>\n\n    Args:\n        questions (`str` or `List[str]`):\n            The questions to be encoded. You can specify one question for many passages. In this case, the question\n            will be duplicated like `[questions] * n_passages`. Otherwise you have to specify as many questions as in\n            `titles` or `texts`.\n        titles (`str` or `List[str]`):\n            The passages titles to be encoded. This can be a string or a list of strings if there are several passages.\n        texts (`str` or `List[str]`):\n            The passages texts to be encoded. This can be a string or a list of strings if there are several passages.\n        padding (`bool`, `str` or [`~utils.PaddingStrategy`], *optional*, defaults to `False`):\n            Activates and controls padding. Accepts the following values:\n\n            - `True` or `'longest'`: Pad to the longest sequence in the batch (or no padding if only a single sequence\n              if provided).\n            - `'max_length'`: Pad to a maximum length specified with the argument `max_length` or to the maximum\n              acceptable input length for the model if that argument is not provided.\n            - `False` or `'do_not_pad'` (default): No padding (i.e., can output a batch with sequences of different\n              lengths).\n        truncation (`bool`, `str` or [`~tokenization_utils_base.TruncationStrategy`], *optional*, defaults to `False`):\n            Activates and controls truncation. Accepts the following values:\n\n            - `True` or `'longest_first'`: Truncate to a maximum length specified with the argument `max_length` or to\n              the maximum acceptable input length for the model if that argument is not provided. This will truncate\n              token by token, removing a token from the longest sequence in the pair if a pair of sequences (or a batch\n              of pairs) is provided.\n            - `'only_first'`: Truncate to a maximum length specified with the argument `max_length` or to the maximum\n              acceptable input length for the model if that argument is not provided. This will only truncate the first\n              sequence of a pair if a pair of sequences (or a batch of pairs) is provided.\n            - `'only_second'`: Truncate to a maximum length specified with the argument `max_length` or to the maximum\n              acceptable input length for the model if that argument is not provided. This will only truncate the\n              second sequence of a pair if a pair of sequences (or a batch of pairs) is provided.\n            - `False` or `'do_not_truncate'` (default): No truncation (i.e., can output batch with sequence lengths\n              greater than the model maximum admissible input size).\n        max_length (`int`, *optional*):\n                Controls the maximum length to use by one of the truncation/padding parameters.\n\n                If left unset or set to `None`, this will use the predefined model maximum length if a maximum length\n                is required by one of the truncation/padding parameters. If the model has no specific maximum input\n                length (like XLNet) truncation/padding to a maximum length will be deactivated.\n        return_tensors (`str` or [`~utils.TensorType`], *optional*):\n                If set, will return tensors instead of list of python integers. Acceptable values are:\n\n                - `'tf'`: Return TensorFlow `tf.constant` objects.\n                - `'pt'`: Return PyTorch `torch.Tensor` objects.\n                - `'np'`: Return Numpy `np.ndarray` objects.\n        return_attention_mask (`bool`, *optional*):\n            Whether or not to return the attention mask. If not set, will return the attention mask according to the\n            specific tokenizer's default, defined by the `return_outputs` attribute.\n\n            [What are attention masks?](../glossary#attention-mask)\n\n    Return:\n        `Dict[str, List[List[int]]]`: A dictionary with the following keys:\n\n        - `input_ids`: List of token ids to be fed to a model.\n        - `attention_mask`: List of indices specifying which tokens should be attended to by the model.\n    "

@add_start_docstrings(_lowerCAmelCase				)
class     a		:




							"""simple docstring"""


							def __call__(							self	:  Union[str, Any]   ,     lowerCamelCase	:  List[str]   ,     lowerCamelCase	:  Optional[str] = None   ,     lowerCamelCase	:  Optional[str] = None   ,     lowerCamelCase	:  Union[bool, str] = False   ,     lowerCamelCase	:  Union[bool, str] = False   ,     lowerCamelCase	:  Optional[int] = None   ,     lowerCamelCase	:  Optional[Union[str, TensorType]] = None   ,     lowerCamelCase	:  Optional[bool] = None   ,     **lowerCamelCase	:  List[str]   ,     )     ->   BatchEncoding:
														if titles is None and texts is None:
																					return super().__call__(
																					    lowerCamelCase   ,     padding=lowerCamelCase   ,     truncation=lowerCamelCase   ,     max_length=lowerCamelCase   ,     return_tensors=lowerCamelCase   ,     return_attention_mask=lowerCamelCase   ,     **lowerCamelCase   ,     )
														elif titles is None or texts is None:
																					__snake_case		:      Any										=       titles if texts is None else texts
																					return super().__call__(
																					    lowerCamelCase   ,     lowerCamelCase   ,     padding=lowerCamelCase   ,     truncation=lowerCamelCase   ,     max_length=lowerCamelCase   ,     return_tensors=lowerCamelCase   ,     return_attention_mask=lowerCamelCase   ,     **lowerCamelCase   ,     )
														__snake_case		:      Any										=       titles if not isinstance(lowerCamelCase   ,     lowerCamelCase    ) else [titles]
														__snake_case		:      Tuple										=       texts if not isinstance(lowerCamelCase   ,     lowerCamelCase    ) else [texts]
														__snake_case		:      Tuple										=       len(lowerCamelCase    )
														__snake_case		:      Optional[int]										=       questions if not isinstance(lowerCamelCase   ,     lowerCamelCase    ) else [questions] * n_passages
														assert len(lowerCamelCase    ) == len(
														    lowerCamelCase    ), F'There should be as many titles than texts but got {len(lowerCamelCase    )} titles and {len(lowerCamelCase    )} texts.'
														__snake_case		:      Dict										=       super().__call__(lowerCamelCase   ,     lowerCamelCase   ,     padding=lowerCamelCase   ,     truncation=lowerCamelCase    )["input_ids"]
														__snake_case		:      List[str]										=       super().__call__(lowerCamelCase   ,     add_special_tokens=lowerCamelCase   ,     padding=lowerCamelCase   ,     truncation=lowerCamelCase    )["input_ids"]
														__snake_case		:      Dict										=       {
														    "input_ids": [
														        (encoded_question_and_title + encoded_text)[:max_length]
														        if max_length is not None and truncation
														        else encoded_question_and_title + encoded_text
														        for encoded_question_and_title, encoded_text in zip(lowerCamelCase   ,     lowerCamelCase    )
														    ]
														}
														if return_attention_mask is not False:
																					__snake_case		:      Optional[int]										=       []
																					for input_ids in encoded_inputs["input_ids"]:
																												attention_mask.append([int(input_id != self.pad_token_id    ) for input_id in input_ids]    )
																					__snake_case		:      Any										=       attention_mask
														return self.pad(lowerCamelCase   ,     padding=lowerCamelCase   ,     max_length=lowerCamelCase   ,     return_tensors=lowerCamelCase    )

							def    __snake_case (							self	:  Union[str, Any]   ,     lowerCamelCase	:  BatchEncoding   ,     lowerCamelCase	:  DPRReaderOutput   ,     lowerCamelCase	:  int = 16   ,     lowerCamelCase	:  int = 64   ,     lowerCamelCase	:  int = 4   ,     )     ->   List[DPRSpanPrediction]:
														__snake_case		:      Any										=       reader_input["input_ids"]
														__snake_case			,      __snake_case			,      __snake_case		:      Optional[int]										=       reader_output[:3]
														__snake_case		:      Union[str, Any]										=       len(lowerCamelCase    )
														__snake_case		:      Any										=       sorted(range(lowerCamelCase    )   ,     reverse=lowerCamelCase   ,     key=relevance_logits.__getitem__    )
														__snake_case		:      List[DPRReaderOutput]										=       []
														for doc_id in sorted_docs:
																					__snake_case		:      List[str]										=       list(input_ids[doc_id]    )
																					# assuming question & title information is at the beginning of the sequence
																					__snake_case		:      int										=       sequence_ids.index(self.sep_token_id   ,     2    ) + 1  # second sep id
																					if sequence_ids[-1] == self.pad_token_id:
																												__snake_case		:      Optional[int]										=       sequence_ids.index(self.pad_token_id    )
																					else:
																												__snake_case		:      Tuple										=       len(lowerCamelCase    )

																					__snake_case		:      str										=       self._get_best_spans(
																					    start_logits=start_logits[doc_id][passage_offset:sequence_len]   ,     end_logits=end_logits[doc_id][passage_offset:sequence_len]   ,     max_answer_length=lowerCamelCase   ,     top_spans=lowerCamelCase   ,     )
																					for start_index, end_index in best_spans:
																												start_index += passage_offset
																												end_index += passage_offset
																												nbest_spans_predictions.append(
																												    DPRSpanPrediction(
																												        span_score=start_logits[doc_id][start_index] + end_logits[doc_id][end_index]   ,     relevance_score=relevance_logits[doc_id]   ,     doc_id=lowerCamelCase   ,     start_index=lowerCamelCase   ,     end_index=lowerCamelCase   ,     text=self.decode(sequence_ids[start_index : end_index + 1]    )   ,     )    )
																					if len(lowerCamelCase    ) >= num_spans:
																												break
														return nbest_spans_predictions[:num_spans]






							def    __snake_case (							self	:  str   ,     lowerCamelCase	:  List[int]   ,     lowerCamelCase	:  List[int]   ,     lowerCamelCase	:  int   ,     lowerCamelCase	:  int   ,     )     ->   List[DPRSpanPrediction]:
														__snake_case		:      Any										=       []
														for start_index, start_score in enumerate(lowerCamelCase    ):
																					for answer_length, end_score in enumerate(end_logits[start_index : start_index + max_answer_length]    ):
																												scores.append(((start_index, start_index + answer_length), start_score + end_score)    )
														__snake_case		:      str										=       sorted(lowerCamelCase   ,     key=lambda lowerCamelCase    : x[1]   ,     reverse=lowerCamelCase    )
														__snake_case		:      int										=       []
														for (start_index, end_index), score in scores:
																					assert start_index <= end_index, F'Wrong span indices: [{start_index}:{end_index}]'
																					__snake_case		:      str										=       end_index - start_index + 1
																					assert length <= max_answer_length, F'Span is too long: {length} > {max_answer_length}'
																					if any(
																					    start_index <= prev_start_index <= prev_end_index <= end_index
																					    or prev_start_index <= start_index <= end_index <= prev_end_index
																					    for (prev_start_index, prev_end_index) in chosen_span_intervals    ):
																												continue
																					chosen_span_intervals.append((start_index, end_index)    )

																					if len(lowerCamelCase    ) == top_spans:
																												break
														return chosen_span_intervals

@add_end_docstrings(_lowerCAmelCase				)
class     a		(_lowerCAmelCase					,    _lowerCAmelCase				):




							"""simple docstring"""


							__UpperCAmelCase   :							Any					  =  VOCAB_FILES_NAMES
							__UpperCAmelCase   :							Union[str, Any]					  =  READER_PRETRAINED_VOCAB_FILES_MAP
							__UpperCAmelCase   :							Any					  =  READER_PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
							__UpperCAmelCase   :							Optional[Any]					  =  READER_PRETRAINED_INIT_CONFIGURATION
							__UpperCAmelCase   :							Union[str, Any]					  =  ["input_ids", "attention_mask"]
							__UpperCAmelCase   :							str					  =  DPRReaderTokenizer





from typing import List, Optional, Union

import torch
from transformers import (
    XLMRobertaTokenizer,
)

from ...models import UNetaDConditionModel, VQModel
from ...pipelines import DiffusionPipeline
from ...pipelines.pipeline_utils import ImagePipelineOutput
from ...schedulers import DDIMScheduler, DDPMScheduler
from ...utils import (
    is_accelerate_available,
    is_accelerate_version,
    logging,
    randn_tensor,
    replace_example_docstring,
)
from .text_encoder import MultilingualCLIP


_snake_case					:		Optional[int]      =    logging.get_logger(__name__)  # pylint: disable=invalid-name

_snake_case					:		Union[str, Any]      =    "\n    Examples:\n        ```py\n        >>> from diffusers import KandinskyPipeline, KandinskyPriorPipeline\n        >>> import torch\n\n        >>> pipe_prior = KandinskyPriorPipeline.from_pretrained(\"kandinsky-community/Kandinsky-2-1-prior\")\n        >>> pipe_prior.to(\"cuda\")\n\n        >>> prompt = \"red cat, 4k photo\"\n        >>> out = pipe_prior(prompt)\n        >>> image_emb = out.image_embeds\n        >>> negative_image_emb = out.negative_image_embeds\n\n        >>> pipe = KandinskyPipeline.from_pretrained(\"kandinsky-community/kandinsky-2-1\")\n        >>> pipe.to(\"cuda\")\n\n        >>> image = pipe(\n        ...     prompt,\n        ...     image_embeds=image_emb,\n        ...     negative_image_embeds=negative_image_emb,\n        ...     height=768,\n        ...     width=768,\n        ...     num_inference_steps=100,\n        ... ).images\n\n        >>> image[0].save(\"cat.png\")\n        ```\n"







def     lowerCAmelCase_					( __lowerCamelCase							,     __lowerCamelCase							,     __lowerCamelCase=8     ):
							__snake_case		:      List[Any]										=       h // scale_factor**2
							if h % scale_factor**2 != 0:
														new_h += 1
							__snake_case		:      Optional[int]										=       w // scale_factor**2
							if w % scale_factor**2 != 0:
														new_w += 1
							return new_h * scale_factor, new_w * scale_factor

class     a		(_lowerCAmelCase				):




							"""simple docstring"""


							def __init__(							self	:  Union[str, Any]   ,     lowerCamelCase	:  MultilingualCLIP   ,     lowerCamelCase	:  XLMRobertaTokenizer   ,     lowerCamelCase	:  UNetaDConditionModel   ,     lowerCamelCase	:  Union[DDIMScheduler, DDPMScheduler]   ,     lowerCamelCase	:  VQModel   ,     )     ->   Optional[int]:
														super().__init__()

														self.register_modules(
														    text_encoder=lowerCamelCase   ,     tokenizer=lowerCamelCase   ,     unet=lowerCamelCase   ,     scheduler=lowerCamelCase   ,     movq=lowerCamelCase   ,     )
														__snake_case		:      List[Any]										=       2 ** (len(self.movq.config.block_out_channels    ) - 1)

							def    __snake_case (							self	:  Any   ,     lowerCamelCase	:  Dict   ,     lowerCamelCase	:  List[Any]   ,     lowerCamelCase	:  Optional[int]   ,     lowerCamelCase	:  Union[str, Any]   ,     lowerCamelCase	:  Optional[Any]   ,     lowerCamelCase	:  int    )     ->   Any:
														if latents is None:
																					__snake_case		:      str										=       randn_tensor(lowerCamelCase   ,     generator=lowerCamelCase   ,     device=lowerCamelCase   ,     dtype=lowerCamelCase    )
														else:
																					if latents.shape != shape:
																												raise ValueError(F'Unexpected latents shape, got {latents.shape}, expected {shape}'    )
																					__snake_case		:      Optional[int]										=       latents.to(lowerCamelCase    )

														__snake_case		:      List[Any]										=       latents * scheduler.init_noise_sigma
														return latents

							def    __snake_case (							self	:  Optional[int]   ,     lowerCamelCase	:  int   ,     lowerCamelCase	:  Any   ,     lowerCamelCase	:  Any   ,     lowerCamelCase	:  List[Any]   ,     lowerCamelCase	:  str=None   ,     )     ->   List[str]:
														__snake_case		:      Tuple										=       len(lowerCamelCase    ) if isinstance(lowerCamelCase   ,     lowerCamelCase    ) else 1
														# get prompt text embeddings
														__snake_case		:      Optional[int]										=       self.tokenizer(
														    lowerCamelCase   ,     padding="max_length"   ,     truncation=lowerCamelCase   ,     max_length=77   ,     return_attention_mask=lowerCamelCase   ,     add_special_tokens=lowerCamelCase   ,     return_tensors="pt"   ,     )

														__snake_case		:      List[str]										=       text_inputs.input_ids
														__snake_case		:      List[Any]										=       self.tokenizer(lowerCamelCase   ,     padding="longest"   ,     return_tensors="pt"    ).input_ids

														if untruncated_ids.shape[-1] >= text_input_ids.shape[-1] and not torch.equal(lowerCamelCase   ,     lowerCamelCase    ):
																					__snake_case		:      Optional[Any]										=       self.tokenizer.batch_decode(untruncated_ids[:, self.tokenizer.model_max_length - 1 : -1]    )
																					logger.warning(
																					    "The following part of your input was truncated because CLIP can only handle sequences up to"
																					    F' {self.tokenizer.model_max_length} tokens: {removed_text}'    )

														__snake_case		:      Any										=       text_input_ids.to(lowerCamelCase    )
														__snake_case		:      List[str]										=       text_inputs.attention_mask.to(lowerCamelCase    )

														__snake_case			,      __snake_case		:      List[str]										=       self.text_encoder(
														    input_ids=lowerCamelCase   ,     attention_mask=lowerCamelCase    )

														__snake_case		:      List[Any]										=       prompt_embeds.repeat_interleave(lowerCamelCase   ,     dim=0    )
														__snake_case		:      List[str]										=       text_encoder_hidden_states.repeat_interleave(lowerCamelCase   ,     dim=0    )
														__snake_case		:      Optional[int]										=       text_mask.repeat_interleave(lowerCamelCase   ,     dim=0    )

														if do_classifier_free_guidance:
																					__snake_case		:      List[str]
																					if negative_prompt is None:
																												__snake_case		:      Any										=       [""] * batch_size
																					elif type(lowerCamelCase    ) is not type(lowerCamelCase    ):
																												raise TypeError(
																												    F'`negative_prompt` should be the same type to `prompt`, but got {type(lowerCamelCase    )} !='
																												    F' {type(lowerCamelCase    )}.'    )
																					elif isinstance(lowerCamelCase   ,     lowerCamelCase    ):
																												__snake_case		:      List[Any]										=       [negative_prompt]
																					elif batch_size != len(lowerCamelCase    ):
																												raise ValueError(
																												    F'`negative_prompt`: {negative_prompt} has batch size {len(lowerCamelCase    )}, but `prompt`:'
																												    F' {prompt} has batch size {batch_size}. Please make sure that passed `negative_prompt` matches'
																												    " the batch size of `prompt`."    )
																					else:
																												__snake_case		:      int										=       negative_prompt

																					__snake_case		:      Dict										=       self.tokenizer(
																					    lowerCamelCase   ,     padding="max_length"   ,     max_length=77   ,     truncation=lowerCamelCase   ,     return_attention_mask=lowerCamelCase   ,     add_special_tokens=lowerCamelCase   ,     return_tensors="pt"   ,     )
																					__snake_case		:      Dict										=       uncond_input.input_ids.to(lowerCamelCase    )
																					__snake_case		:      List[Any]										=       uncond_input.attention_mask.to(lowerCamelCase    )

																					__snake_case			,      __snake_case		:      Tuple										=       self.text_encoder(
																					    input_ids=lowerCamelCase   ,     attention_mask=lowerCamelCase    )

																					# duplicate unconditional embeddings for each generation per prompt, using mps friendly method

																					__snake_case		:      Dict										=       negative_prompt_embeds.shape[1]
																					__snake_case		:      int										=       negative_prompt_embeds.repeat(1   ,     lowerCamelCase    )
																					__snake_case		:      List[str]										=       negative_prompt_embeds.view(batch_size * num_images_per_prompt   ,     lowerCamelCase    )

																					__snake_case		:      Union[str, Any]										=       uncond_text_encoder_hidden_states.shape[1]
																					__snake_case		:      Tuple										=       uncond_text_encoder_hidden_states.repeat(1   ,     lowerCamelCase   ,     1    )
																					__snake_case		:      str										=       uncond_text_encoder_hidden_states.view(
																					    batch_size * num_images_per_prompt   ,     lowerCamelCase   ,     -1    )
																					__snake_case		:      Optional[int]										=       uncond_text_mask.repeat_interleave(lowerCamelCase   ,     dim=0    )

																					# done duplicates

																					# For classifier free guidance, we need to do two forward passes.
																					# Here we concatenate the unconditional and text embeddings into a single batch
																					# to avoid doing two forward passes
																					__snake_case		:      Optional[int]										=       torch.cat([negative_prompt_embeds, prompt_embeds]    )
																					__snake_case		:      List[Any]										=       torch.cat([uncond_text_encoder_hidden_states, text_encoder_hidden_states]    )

																					__snake_case		:      Any										=       torch.cat([uncond_text_mask, text_mask]    )

														return prompt_embeds, text_encoder_hidden_states, text_mask

							def    __snake_case (							self	:  List[str]   ,     lowerCamelCase	:  Dict=0    )     ->   Tuple:
														if is_accelerate_available():
																					from accelerate import cpu_offload
														else:
																					raise ImportError("Please install accelerate via `pip install accelerate`"    )

														__snake_case		:      Optional[int]										=       torch.device(F'cuda:{gpu_id}'    )

														__snake_case		:      Optional[Any]										=       [
														    self.unet,
														    self.text_encoder,
														    self.movq,
														]
														for cpu_offloaded_model in models:
																					if cpu_offloaded_model is not None:
																												cpu_offload(lowerCamelCase   ,     lowerCamelCase    )

							def    __snake_case (							self	:  List[Any]   ,     lowerCamelCase	:  int=0    )     ->   Optional[int]:
														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."    )

														__snake_case		:      Optional[Any]										=       torch.device(F'cuda:{gpu_id}'    )

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

														__snake_case		:      List[str]										=       None
														for cpu_offloaded_model in [self.text_encoder, self.unet, self.movq]:
																					__snake_case			,      __snake_case		:      List[Any]										=       cpu_offload_with_hook(lowerCamelCase   ,     lowerCamelCase   ,     prev_module_hook=lowerCamelCase    )

														if self.safety_checker is not None:
																					__snake_case			,      __snake_case		:      Optional[int]										=       cpu_offload_with_hook(self.safety_checker   ,     lowerCamelCase   ,     prev_module_hook=lowerCamelCase    )

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

							@property
							# Copied from diffusers.pipelines.stable_diffusion.pipeline_stable_diffusion.StableDiffusionPipeline._execution_device
							def    __snake_case (							self	:  List[Any]    )     ->   Optional[int]:
														if not hasattr(self.unet   ,     "_hf_hook"    ):
																					return self.device
														for module in self.unet.modules():
																					if (
																					    hasattr(lowerCamelCase   ,     "_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






							@torch.no_grad()
							@replace_example_docstring(lowerCamelCase    )
							def __call__(							self	:  Dict   ,     lowerCamelCase	:  Union[str, List[str]]   ,     lowerCamelCase	:  Union[torch.FloatTensor, List[torch.FloatTensor]]   ,     lowerCamelCase	:  Union[torch.FloatTensor, List[torch.FloatTensor]]   ,     lowerCamelCase	:  Optional[Union[str, List[str]]] = None   ,     lowerCamelCase	:  int = 512   ,     lowerCamelCase	:  int = 512   ,     lowerCamelCase	:  int = 100   ,     lowerCamelCase	:  float = 4.0   ,     lowerCamelCase	:  int = 1   ,     lowerCamelCase	:  Optional[Union[torch.Generator, List[torch.Generator]]] = None   ,     lowerCamelCase	:  Optional[torch.FloatTensor] = None   ,     lowerCamelCase	:  Optional[str] = "pil"   ,     lowerCamelCase	:  bool = True   ,     )     ->   List[Any]:

														if isinstance(lowerCamelCase   ,     lowerCamelCase    ):
																					__snake_case		:      Optional[int]										=       1
														elif isinstance(lowerCamelCase   ,     lowerCamelCase    ):
																					__snake_case		:      List[Any]										=       len(lowerCamelCase    )
														else:
																					raise ValueError(F'`prompt` has to be of type `str` or `list` but is {type(lowerCamelCase    )}'    )

														__snake_case		:      Any										=       self._execution_device

														__snake_case		:      Any										=       batch_size * num_images_per_prompt
														__snake_case		:      Any										=       guidance_scale > 1.0

														__snake_case			,      __snake_case			,      __snake_case		:      Optional[Any]										=       self._encode_prompt(
														    lowerCamelCase   ,     lowerCamelCase   ,     lowerCamelCase   ,     lowerCamelCase   ,     lowerCamelCase    )

														if isinstance(lowerCamelCase   ,     lowerCamelCase    ):
																					__snake_case		:      List[Any]										=       torch.cat(lowerCamelCase   ,     dim=0    )
														if isinstance(lowerCamelCase   ,     lowerCamelCase    ):
																					__snake_case		:      str										=       torch.cat(lowerCamelCase   ,     dim=0    )

														if do_classifier_free_guidance:
																					__snake_case		:      Dict										=       image_embeds.repeat_interleave(lowerCamelCase   ,     dim=0    )
																					__snake_case		:      Optional[Any]										=       negative_image_embeds.repeat_interleave(lowerCamelCase   ,     dim=0    )

														__snake_case		:      str										=       torch.cat([negative_image_embeds, image_embeds]   ,     dim=0    ).to(
														    dtype=prompt_embeds.dtype   ,     device=lowerCamelCase    )

														self.scheduler.set_timesteps(lowerCamelCase   ,     device=lowerCamelCase    )
														__snake_case		:      Tuple										=       self.scheduler.timesteps

														__snake_case		:      Union[str, Any]										=       self.unet.config.in_channels

														__snake_case			,      __snake_case		:      Tuple										=       get_new_h_w(lowerCamelCase   ,     lowerCamelCase   ,     self.movq_scale_factor    )

														# create initial latent
														__snake_case		:      Any										=       self.prepare_latents(
														    (batch_size, num_channels_latents, height, width)   ,     text_encoder_hidden_states.dtype   ,     lowerCamelCase   ,     lowerCamelCase   ,     lowerCamelCase   ,     self.scheduler   ,     )

														for i, t in enumerate(self.progress_bar(lowerCamelCase    )    ):
																					# expand the latents if we are doing classifier free guidance
																					__snake_case		:      Any										=       torch.cat([latents] * 2    ) if do_classifier_free_guidance else latents

																					__snake_case		:      int										=       {"text_embeds": prompt_embeds, "image_embeds": image_embeds}
																					__snake_case		:      Optional[Any]										=       self.unet(
																					    sample=lowerCamelCase   ,     timestep=lowerCamelCase   ,     encoder_hidden_states=lowerCamelCase   ,     added_cond_kwargs=lowerCamelCase   ,     return_dict=lowerCamelCase   ,     )[0]

																					if do_classifier_free_guidance:
																												__snake_case			,      __snake_case		:      Any										=       noise_pred.split(latents.shape[1]   ,     dim=1    )
																												__snake_case			,      __snake_case		:      Union[str, Any]										=       noise_pred.chunk(2    )
																												__snake_case			,      __snake_case		:      str										=       variance_pred.chunk(2    )
																												__snake_case		:      Optional[Any]										=       noise_pred_uncond + guidance_scale * (noise_pred_text - noise_pred_uncond)
																												__snake_case		:      Union[str, Any]										=       torch.cat([noise_pred, variance_pred_text]   ,     dim=1    )

																					if not (
																					    hasattr(self.scheduler.config   ,     "variance_type"    )
																					    and self.scheduler.config.variance_type in ["learned", "learned_range"]
																					):
																												__snake_case			,      __snake_case		:      Union[str, Any]										=       noise_pred.split(latents.shape[1]   ,     dim=1    )

																					# compute the previous noisy sample x_t -> x_t-1
																					__snake_case		:      str										=       self.scheduler.step(
																					    lowerCamelCase   ,     lowerCamelCase   ,     lowerCamelCase   ,     generator=lowerCamelCase   ,     ).prev_sample
														# post-processing
														__snake_case		:      str										=       self.movq.decode(lowerCamelCase   ,     force_not_quantize=lowerCamelCase    )["sample"]

														if output_type not in ["pt", "np", "pil"]:
																					raise ValueError(F'Only the output types `pt`, `pil` and `np` are supported not output_type={output_type}'    )

														if output_type in ["np", "pil"]:
																					__snake_case		:      Union[str, Any]										=       image * 0.5 + 0.5
																					__snake_case		:      Union[str, Any]										=       image.clamp(0   ,     1    )
																					__snake_case		:      Optional[int]										=       image.cpu().permute(0   ,     2   ,     3   ,     1    ).float().numpy()

														if output_type == "pil":
																					__snake_case		:      str										=       self.numpy_to_pil(lowerCamelCase    )

														if not return_dict:
																					return (image,)

														return ImagePipelineOutput(images=lowerCamelCase    )

