
import json
import os
import re
import unicodedata
from typing import Optional, Tuple

from ...tokenization_utils import PreTrainedTokenizer, _is_control, _is_punctuation, _is_whitespace
from ...utils import logging


UpperCAmelCase__				:  Optional[Any]       					=				logging.get_logger(__name__)

UpperCAmelCase__				:  List[str]       					=				{
    '''vocab_file''': '''vocab.json''',
    '''merges_file''': '''merges.txt''',
}

UpperCAmelCase__				:  Optional[int]       					=				{
    '''vocab_file''': {'''openai-gpt''': '''https://huggingface.co/openai-gpt/resolve/main/vocab.json'''},
    '''merges_file''': {'''openai-gpt''': '''https://huggingface.co/openai-gpt/resolve/main/merges.txt'''},
}

UpperCAmelCase__				:  Any       					=				{
    '''openai-gpt''': 512,
}



def       _lowercase					(      __SCREAMING_SNAKE_CASE )	->     Optional[int]:
			UpperCamelCase__    :		List[Any]     =				text.strip()
			if not text:
						return []
			UpperCamelCase__    :		List[Any]     =				text.split()
			return tokens



class 			UpperCamelCase_			(						__UpperCamelCase   ):




							'''simple docstring'''

							def __init__(				self     ,						UpperCamelCase=True     ,						UpperCamelCase=None     ,						UpperCamelCase=True     ,						UpperCamelCase=None     ,						UpperCamelCase=True     ,						) ->	Tuple:
										if never_split is None:
													UpperCamelCase__    :		str     =				[]
										UpperCamelCase__    :		Optional[Any]     =				do_lower_case
										UpperCamelCase__    :		Optional[int]     =				set(UpperCamelCase)
										UpperCamelCase__    :		List[Any]     =				tokenize_chinese_chars
										UpperCamelCase__    :		Optional[int]     =				strip_accents
										UpperCamelCase__    :		Any     =				do_split_on_punc
							def        lowerCAmelCase__ (				self     ,						UpperCamelCase     ,						UpperCamelCase=None) ->	List[str]:
										UpperCamelCase__    :		Tuple     =				self.never_split.union(set(UpperCamelCase)) if never_split else self.never_split
										UpperCamelCase__    :		Optional[int]     =				self._clean_text(UpperCamelCase)

										# This was added on November 1st, 2018 for the multilingual and Chinese
										# models. This is also applied to the English models now, but it doesn't
										# matter since the English models were not trained on any Chinese data
										# and generally don't have any Chinese data in them (there are Chinese
										# characters in the vocabulary because Wikipedia does have some Chinese
										# words in the English Wikipedia.).
										if self.tokenize_chinese_chars:
													UpperCamelCase__    :		List[str]     =				self._tokenize_chinese_chars(UpperCamelCase)
										# prevents treating the same character with different unicode codepoints as different characters
										UpperCamelCase__    :		Dict     =				unicodedata.normalize('NFC'     ,						UpperCamelCase)
										UpperCamelCase__    :		Optional[int]     =				whitespace_tokenize(UpperCamelCase)
										UpperCamelCase__    :		List[Any]     =				[]
										for token in orig_tokens:
													if token not in never_split:
																if self.do_lower_case:
																			UpperCamelCase__    :		str     =				token.lower()
																			if self.strip_accents is not False:
																						UpperCamelCase__    :		str     =				self._run_strip_accents(UpperCamelCase)
																elif self.strip_accents:
																			UpperCamelCase__    :		Union[str, Any]     =				self._run_strip_accents(UpperCamelCase)
													split_tokens.extend(self._run_split_on_punc(UpperCamelCase     ,						UpperCamelCase))

										UpperCamelCase__    :		Dict     =				whitespace_tokenize(' '.join(UpperCamelCase))
										return output_tokens
							def        lowerCAmelCase__ (				self     ,						UpperCamelCase) ->	List[str]:
										UpperCamelCase__    :		Union[str, Any]     =				unicodedata.normalize('NFD'     ,						UpperCamelCase)
										UpperCamelCase__    :		Dict     =				[]
										for char in text:
													UpperCamelCase__    :		Union[str, Any]     =				unicodedata.category(UpperCamelCase)
													if cat == "Mn":
																continue
													output.append(UpperCamelCase)
										return "".join(UpperCamelCase)
							def        lowerCAmelCase__ (				self     ,						UpperCamelCase     ,						UpperCamelCase=None) ->	Dict:
										if not self.do_split_on_punc or (never_split is not None and text in never_split):
													return [text]
										UpperCamelCase__    :		Dict     =				list(UpperCamelCase)
										UpperCamelCase__    :		Optional[int]     =				0
										UpperCamelCase__    :		Dict     =				True
										UpperCamelCase__    :		Optional[int]     =				[]
										while i < len(UpperCamelCase):
													UpperCamelCase__    :		List[str]     =				chars[i]
													if _is_punctuation(UpperCamelCase):
																output.append([char])
																UpperCamelCase__    :		Tuple     =				True
													else:
																if start_new_word:
																			output.append([])
																UpperCamelCase__    :		Union[str, Any]     =				False
																output[-1].append(UpperCamelCase)
													i += 1

										return ["".join(UpperCamelCase) for x in output]
							def        lowerCAmelCase__ (				self     ,						UpperCamelCase) ->	Tuple:
										UpperCamelCase__    :		List[str]     =				[]
										for char in text:
													UpperCamelCase__    :		List[str]     =				ord(UpperCamelCase)
													if self._is_chinese_char(UpperCamelCase):
																output.append(' ')
																output.append(UpperCamelCase)
																output.append(' ')
													else:
																output.append(UpperCamelCase)
										return "".join(UpperCamelCase)
							def        lowerCAmelCase__ (				self     ,						UpperCamelCase) ->	Union[str, Any]:
										# This defines a "chinese character" as anything in the CJK Unicode block:
										#   https://en.wikipedia.org/wiki/CJK_Unified_Ideographs_(Unicode_block)
										#
										# Note that the CJK Unicode block is NOT all Japanese and Korean characters,
										# despite its name. The modern Korean Hangul alphabet is a different block,
										# as is Japanese Hiragana and Katakana. Those alphabets are used to write
										# space-separated words, so they are not treated specially and handled
										# like the all of the other languages.
										if (
										    (cp >= 0X4e00 and cp <= 0X9fff)
										    or (cp >= 0X3400 and cp <= 0X4dbf)  #
										    or (cp >= 0X2_0000 and cp <= 0X2_a6df)  #
										    or (cp >= 0X2_a700 and cp <= 0X2_b73f)  #
										    or (cp >= 0X2_b740 and cp <= 0X2_b81f)  #
										    or (cp >= 0X2_b820 and cp <= 0X2_ceaf)  #
										    or (cp >= 0Xf900 and cp <= 0Xfaff)
										    or (cp >= 0X2_f800 and cp <= 0X2_fa1f)  #
										):  #
													return True

										return False






							def        lowerCAmelCase__ (				self     ,						UpperCamelCase) ->	List[str]:
										UpperCamelCase__    :		List[Any]     =				[]
										for char in text:
													UpperCamelCase__    :		Dict     =				ord(UpperCamelCase)
													if cp == 0 or cp == 0Xfffd or _is_control(UpperCamelCase):
																continue
													if _is_whitespace(UpperCamelCase):
																output.append(' ')
													else:
																output.append(UpperCamelCase)
										return "".join(UpperCamelCase)



def       _lowercase					(      __SCREAMING_SNAKE_CASE )	->     Union[str, Any]:
			UpperCamelCase__    :		Optional[int]     =				set()
			UpperCamelCase__    :		str     =				word[0]
			for char in word[1:]:
						pairs.add((prev_char, char) )
						UpperCamelCase__    :		Any     =				char
			return pairs



def       _lowercase					(      __SCREAMING_SNAKE_CASE )	->     Optional[Any]:
			UpperCamelCase__    :		Tuple     =				text.replace('—'				,    '-' )
			UpperCamelCase__    :		Any     =				text.replace('–'				,    '-' )
			UpperCamelCase__    :		Any     =				text.replace('―'				,    '-' )
			UpperCamelCase__    :		Tuple     =				text.replace('…'				,    '...' )
			UpperCamelCase__    :		Dict     =				text.replace('´'				,    '\'' )
			UpperCamelCase__    :		int     =				re.sub(R'(-+|~+|!+|"+|;+|\?+|\++|,+|\)+|\(+|\\+|\/+|\*+|\[+|\]+|}+|{+|\|+|_+)'				,    R' \1 '				,    __SCREAMING_SNAKE_CASE )
			UpperCamelCase__    :		int     =				re.sub(R'\s*\n\s*'				,    ' \n '				,    __SCREAMING_SNAKE_CASE )
			UpperCamelCase__    :		str     =				re.sub(R'[^\S\n]+'				,    ' '				,    __SCREAMING_SNAKE_CASE )
			return text.strip()



class 			UpperCamelCase_			(						__UpperCamelCase   ):




							'''simple docstring'''

							UpperCamelCase_					   =							VOCAB_FILES_NAMES
							UpperCamelCase_					   =							PRETRAINED_VOCAB_FILES_MAP
							UpperCamelCase_					   =							PRETRAINED_POSITIONAL_EMBEDDINGS_SIZES
							UpperCamelCase_					   =							["""input_ids""", """attention_mask"""]
							def __init__(				self     ,						UpperCamelCase     ,						UpperCamelCase     ,						UpperCamelCase="<unk>"     ,						**UpperCamelCase) ->	Union[str, Any]:
										super().__init__(unk_token=UpperCamelCase     ,						**UpperCamelCase)

										try:
													import ftfy
													from spacy.lang.en import English

													UpperCamelCase__    :		Optional[int]     =				English()
													UpperCamelCase__    :		Dict     =				_nlp.tokenizer
													UpperCamelCase__    :		List[Any]     =				ftfy.fix_text
										except ImportError:
													logger.warning('ftfy or spacy is not installed using BERT BasicTokenizer instead of SpaCy & ftfy.')
													UpperCamelCase__    :		Dict     =				BasicTokenizer(do_lower_case=UpperCamelCase)
													UpperCamelCase__    :		List[Any]     =				None

										with open(UpperCamelCase     ,						encoding='utf-8') as vocab_handle:
													UpperCamelCase__    :		int     =				json.load(UpperCamelCase)
										UpperCamelCase__    :		Dict     =				{v: k for k, v in self.encoder.items()}
										with open(UpperCamelCase     ,						encoding='utf-8') as merges_handle:
													UpperCamelCase__    :		Optional[int]     =				merges_handle.read().split('\n')[1:-1]
										UpperCamelCase__    :		int     =				[tuple(merge.split()) for merge in merges]
										UpperCamelCase__    :		int     =				dict(zip(UpperCamelCase     ,						range(len(UpperCamelCase))))
										UpperCamelCase__    :		Optional[Any]     =				{}
							@property
							def        lowerCAmelCase__ (				self) ->	Dict:
										return True
							@property
							def        lowerCAmelCase__ (				self) ->	Union[str, Any]:
										return len(self.encoder)
							def        lowerCAmelCase__ (				self) ->	Tuple:
										return dict(self.encoder     ,						**self.added_tokens_encoder)
							def        lowerCAmelCase__ (				self     ,						UpperCamelCase) ->	List[Any]:
										UpperCamelCase__    :		Optional[Any]     =				tuple(token[:-1]) + (token[-1] + '</w>',)
										if token in self.cache:
													return self.cache[token]
										UpperCamelCase__    :		int     =				get_pairs(UpperCamelCase)

										if not pairs:
													return token + "</w>"

										while True:
													UpperCamelCase__    :		str     =				min(UpperCamelCase     ,						key=lambda UpperCamelCase: self.bpe_ranks.get(UpperCamelCase     ,						float('inf')))
													if bigram not in self.bpe_ranks:
																break
													UpperCamelCase__   ,			UpperCamelCase__    :		Optional[int]     =				bigram
													UpperCamelCase__    :		str     =				[]
													UpperCamelCase__    :		List[Any]     =				0
													while i < len(UpperCamelCase):
																try:
																			UpperCamelCase__    :		Union[str, Any]     =				word.index(UpperCamelCase     ,						UpperCamelCase)
																except ValueError:
																			new_word.extend(word[i:])
																			break
																else:
																			new_word.extend(word[i:j])
																			UpperCamelCase__    :		Tuple     =				j

																if word[i] == first and i < len(UpperCamelCase) - 1 and word[i + 1] == second:
																			new_word.append(first + second)
																			i += 2
																else:
																			new_word.append(word[i])
																			i += 1
													UpperCamelCase__    :		int     =				tuple(UpperCamelCase)
													UpperCamelCase__    :		List[Any]     =				new_word
													if len(UpperCamelCase) == 1:
																break
													else:
																UpperCamelCase__    :		List[Any]     =				get_pairs(UpperCamelCase)
										UpperCamelCase__    :		Optional[int]     =				' '.join(UpperCamelCase)
										if word == "\n  </w>":
													UpperCamelCase__    :		Optional[int]     =				'\n</w>'
										UpperCamelCase__    :		Optional[Any]     =				word
										return word
							def        lowerCAmelCase__ (				self     ,						UpperCamelCase) ->	List[Any]:
										UpperCamelCase__    :		int     =				[]
										if self.fix_text is None:
													# Using BERT's BasicTokenizer
													UpperCamelCase__    :		List[str]     =				self.nlp.tokenize(UpperCamelCase)
													for token in text:
																split_tokens.extend(list(self.bpe(UpperCamelCase).split(' ')))
										else:
													# Using SpaCy & ftfy (original tokenization process of OpenAI GPT)
													UpperCamelCase__    :		Dict     =				self.nlp(text_standardize(self.fix_text(UpperCamelCase)))
													for token in text:
																split_tokens.extend(list(self.bpe(token.text.lower()).split(' ')))
										return split_tokens
							def        lowerCAmelCase__ (				self     ,						UpperCamelCase) ->	Tuple:
										return self.encoder.get(UpperCamelCase     ,						self.encoder.get(self.unk_token))
							def        lowerCAmelCase__ (				self     ,						UpperCamelCase) ->	List[Any]:
										return self.decoder.get(UpperCamelCase     ,						self.unk_token)
							def        lowerCAmelCase__ (				self     ,						UpperCamelCase) ->	List[str]:
										UpperCamelCase__    :		Optional[Any]     =				''.join(UpperCamelCase).replace('</w>'     ,						' ').strip()
										return out_string






							def        lowerCAmelCase__ (				self     ,						UpperCamelCase     ,						UpperCamelCase = None) ->	Tuple[str]:
										if not os.path.isdir(UpperCamelCase):
													logger.error(F"""Vocabulary path ({save_directory}) should be a directory""")
													return
										UpperCamelCase__    :		int     =				os.path.join(
										    UpperCamelCase     ,						(filename_prefix + '-' if filename_prefix else '') + VOCAB_FILES_NAMES['vocab_file'])
										UpperCamelCase__    :		List[str]     =				os.path.join(
										    UpperCamelCase     ,						(filename_prefix + '-' if filename_prefix else '') + VOCAB_FILES_NAMES['merges_file'])

										with open(UpperCamelCase     ,						'w'     ,						encoding='utf-8') as f:
													f.write(json.dumps(self.encoder     ,						indent=2     ,						sort_keys=UpperCamelCase     ,						ensure_ascii=UpperCamelCase) + '\n')

										UpperCamelCase__    :		Any     =				0
										with open(UpperCamelCase     ,						'w'     ,						encoding='utf-8') as writer:
													writer.write('#version: 0.2\n')
													for bpe_tokens, token_index in sorted(self.bpe_ranks.items()     ,						key=lambda UpperCamelCase: kv[1]):
																if index != token_index:
																			logger.warning(
																			    F"""Saving vocabulary to {merge_file}: BPE merge indices are not consecutive."""
																			    ' Please check that the tokenizer is not corrupted!')
																			UpperCamelCase__    :		Dict     =				token_index
																writer.write(' '.join(UpperCamelCase) + '\n')
																index += 1

										return vocab_file, merge_file





from typing import List, Optional, Union

import numpy as np
import PIL
import torch
from PIL import Image

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


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

UpperCAmelCase__				:  Optional[int]       					=				'''
    Examples:
        ```py
        >>> import torch
        >>> import numpy as np

        >>> from diffusers import KandinskyV22PriorEmb2EmbPipeline, KandinskyV22ControlnetImg2ImgPipeline
        >>> from transformers import pipeline
        >>> from diffusers.utils import load_image


        >>> def make_hint(image, depth_estimator):
        ...     image = depth_estimator(image)["depth"]
        ...     image = np.array(image)
        ...     image = image[:, :, None]
        ...     image = np.concatenate([image, image, image], axis=2)
        ...     detected_map = torch.from_numpy(image).float() / 255.0
        ...     hint = detected_map.permute(2, 0, 1)
        ...     return hint


        >>> depth_estimator = pipeline("depth-estimation")

        >>> pipe_prior = KandinskyV22PriorEmb2EmbPipeline.from_pretrained(
        ...     "kandinsky-community/kandinsky-2-2-prior", torch_dtype=torch.float16
        ... )
        >>> pipe_prior = pipe_prior.to("cuda")

        >>> pipe = KandinskyV22ControlnetImg2ImgPipeline.from_pretrained(
        ...     "kandinsky-community/kandinsky-2-2-controlnet-depth", torch_dtype=torch.float16
        ... )
        >>> pipe = pipe.to("cuda")

        >>> img = load_image(
        ...     "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"
        ...     "/kandinsky/cat.png"
        ... ).resize((768, 768))


        >>> hint = make_hint(img, depth_estimator).unsqueeze(0).half().to("cuda")

        >>> prompt = "A robot, 4k photo"
        >>> negative_prior_prompt = "lowres, text, error, cropped, worst quality, low quality, jpeg artifacts, ugly, duplicate, morbid, mutilated, out of frame, extra fingers, mutated hands, poorly drawn hands, poorly drawn face, mutation, deformed, blurry, dehydrated, bad anatomy, bad proportions, extra limbs, cloned face, disfigured, gross proportions, malformed limbs, missing arms, missing legs, extra arms, extra legs, fused fingers, too many fingers, long neck, username, watermark, signature"

        >>> generator = torch.Generator(device="cuda").manual_seed(43)

        >>> img_emb = pipe_prior(prompt=prompt, image=img, strength=0.85, generator=generator)
        >>> negative_emb = pipe_prior(prompt=negative_prior_prompt, image=img, strength=1, generator=generator)

        >>> images = pipe(
        ...     image=img,
        ...     strength=0.5,
        ...     image_embeds=img_emb.image_embeds,
        ...     negative_image_embeds=negative_emb.image_embeds,
        ...     hint=hint,
        ...     num_inference_steps=50,
        ...     generator=generator,
        ...     height=768,
        ...     width=768,
        ... ).images

        >>> images[0].save("robot_cat.png")
        ```
'''



def       _lowercase					(      __SCREAMING_SNAKE_CASE				,    __SCREAMING_SNAKE_CASE				,    __SCREAMING_SNAKE_CASE=8 )	->     List[Any]:
			UpperCamelCase__    :		Union[str, Any]     =				height // scale_factor**2
			if height % scale_factor**2 != 0:
						new_height += 1
			UpperCamelCase__    :		Tuple     =				width // scale_factor**2
			if width % scale_factor**2 != 0:
						new_width += 1
			return new_height * scale_factor, new_width * scale_factor



def       _lowercase					(      __SCREAMING_SNAKE_CASE				,    __SCREAMING_SNAKE_CASE=512				,    __SCREAMING_SNAKE_CASE=512 )	->     Union[str, Any]:
			UpperCamelCase__    :		List[str]     =				pil_image.resize((w, h)				,    resample=Image.BICUBIC				,    reducing_gap=1 )
			UpperCamelCase__    :		List[str]     =				np.array(pil_image.convert('RGB' ) )
			UpperCamelCase__    :		Tuple     =				arr.astype(np.floataa ) / 127.5 - 1
			UpperCamelCase__    :		Tuple     =				np.transpose(__SCREAMING_SNAKE_CASE				,    [2, 0, 1] )
			UpperCamelCase__    :		int     =				torch.from_numpy(__SCREAMING_SNAKE_CASE ).unsqueeze(0 )
			return image



class 			UpperCamelCase_			(						__UpperCamelCase   ):




							'''simple docstring'''

							def __init__(				self     ,						UpperCamelCase     ,						UpperCamelCase     ,						UpperCamelCase     ,						) ->	Any:
										super().__init__()

										self.register_modules(
										    unet=UpperCamelCase     ,						scheduler=UpperCamelCase     ,						movq=UpperCamelCase     ,						)
										UpperCamelCase__    :		str     =				2 ** (len(self.movq.config.block_out_channels) - 1)
							def        lowerCAmelCase__ (				self     ,						UpperCamelCase     ,						UpperCamelCase     ,						UpperCamelCase) ->	Tuple:
										# get the original timestep using init_timestep
										UpperCamelCase__    :		str     =				min(int(num_inference_steps * strength)     ,						UpperCamelCase)

										UpperCamelCase__    :		int     =				max(num_inference_steps - init_timestep     ,						0)
										UpperCamelCase__    :		str     =				self.scheduler.timesteps[t_start:]

										return timesteps, num_inference_steps - t_start
							def        lowerCAmelCase__ (				self     ,						UpperCamelCase     ,						UpperCamelCase     ,						UpperCamelCase     ,						UpperCamelCase     ,						UpperCamelCase     ,						UpperCamelCase     ,						UpperCamelCase=None) ->	Optional[Any]:
										if not isinstance(UpperCamelCase     ,						(torch.Tensor, PIL.Image.Image, list)):
													raise ValueError(
													    F"""`image` has to be of type `torch.Tensor`, `PIL.Image.Image` or list but is {type(UpperCamelCase)}""")

										UpperCamelCase__    :		int     =				image.to(device=UpperCamelCase     ,						dtype=UpperCamelCase)

										UpperCamelCase__    :		int     =				batch_size * num_images_per_prompt

										if image.shape[1] == 4:
													UpperCamelCase__    :		Tuple     =				image

										else:
													if isinstance(UpperCamelCase     ,						UpperCamelCase) and len(UpperCamelCase) != batch_size:
																raise ValueError(
																    F"""You have passed a list of generators of length {len(UpperCamelCase)}, but requested an effective batch"""
																    F""" size of {batch_size}. Make sure the batch size matches the length of the generators.""")

													elif isinstance(UpperCamelCase     ,						UpperCamelCase):
																UpperCamelCase__    :		Union[str, Any]     =				[
																    self.movq.encode(image[i : i + 1]).latent_dist.sample(generator[i]) for i in range(UpperCamelCase)
																]
																UpperCamelCase__    :		Any     =				torch.cat(UpperCamelCase     ,						dim=0)
													else:
																UpperCamelCase__    :		Any     =				self.movq.encode(UpperCamelCase).latent_dist.sample(UpperCamelCase)

													UpperCamelCase__    :		List[Any]     =				self.movq.config.scaling_factor * init_latents

										UpperCamelCase__    :		List[Any]     =				torch.cat([init_latents]     ,						dim=0)

										UpperCamelCase__    :		Union[str, Any]     =				init_latents.shape
										UpperCamelCase__    :		Union[str, Any]     =				randn_tensor(UpperCamelCase     ,						generator=UpperCamelCase     ,						device=UpperCamelCase     ,						dtype=UpperCamelCase)

										# get latents
										UpperCamelCase__    :		Optional[int]     =				self.scheduler.add_noise(UpperCamelCase     ,						UpperCamelCase     ,						UpperCamelCase)

										UpperCamelCase__    :		List[Any]     =				init_latents

										return latents
							def        lowerCAmelCase__ (				self     ,						UpperCamelCase=0) ->	str:
										if is_accelerate_available():
													from accelerate import cpu_offload
										else:
													raise ImportError('Please install accelerate via `pip install accelerate`')

										UpperCamelCase__    :		Any     =				torch.device(F"""cuda:{gpu_id}""")

										UpperCamelCase__    :		int     =				[
										    self.unet,
										    self.movq,
										]
										for cpu_offloaded_model in models:
													if cpu_offloaded_model is not None:
																cpu_offload(UpperCamelCase     ,						UpperCamelCase)
							def        lowerCAmelCase__ (				self     ,						UpperCamelCase=0) ->	Tuple:
										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.')

										UpperCamelCase__    :		int     =				torch.device(F"""cuda:{gpu_id}""")

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

										UpperCamelCase__    :		Dict     =				None
										for cpu_offloaded_model in [self.unet, self.movq]:
													UpperCamelCase__   ,			UpperCamelCase__    :		Optional[int]     =				cpu_offload_with_hook(UpperCamelCase     ,						UpperCamelCase     ,						prev_module_hook=UpperCamelCase)

										# We'll offload the last model manually.
										UpperCamelCase__    :		List[Any]     =				hook
							@property
							# Copied from diffusers.pipelines.stable_diffusion.pipeline_stable_diffusion.StableDiffusionPipeline._execution_device
							def        lowerCAmelCase__ (				self) ->	Optional[Any]:
										if not hasattr(self.unet     ,						'_hf_hook'):
													return self.device
										for module in self.unet.modules():
													if (
													    hasattr(UpperCamelCase     ,						'_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(UpperCamelCase)
							def __call__(				self     ,						UpperCamelCase     ,						UpperCamelCase     ,						UpperCamelCase     ,						UpperCamelCase     ,						UpperCamelCase = 5_12     ,						UpperCamelCase = 5_12     ,						UpperCamelCase = 1_00     ,						UpperCamelCase = 4.0     ,						UpperCamelCase = 0.3     ,						UpperCamelCase = 1     ,						UpperCamelCase = None     ,						UpperCamelCase = "pil"     ,						UpperCamelCase = True     ,						) ->	Optional[Any]:
										UpperCamelCase__    :		List[Any]     =				self._execution_device

										UpperCamelCase__    :		Union[str, Any]     =				guidance_scale > 1.0

										if isinstance(UpperCamelCase     ,						UpperCamelCase):
													UpperCamelCase__    :		Dict     =				torch.cat(UpperCamelCase     ,						dim=0)
										if isinstance(UpperCamelCase     ,						UpperCamelCase):
													UpperCamelCase__    :		List[Any]     =				torch.cat(UpperCamelCase     ,						dim=0)
										if isinstance(UpperCamelCase     ,						UpperCamelCase):
													UpperCamelCase__    :		Any     =				torch.cat(UpperCamelCase     ,						dim=0)

										UpperCamelCase__    :		Optional[Any]     =				image_embeds.shape[0]

										if do_classifier_free_guidance:
													UpperCamelCase__    :		List[str]     =				image_embeds.repeat_interleave(UpperCamelCase     ,						dim=0)
													UpperCamelCase__    :		Any     =				negative_image_embeds.repeat_interleave(UpperCamelCase     ,						dim=0)
													UpperCamelCase__    :		str     =				hint.repeat_interleave(UpperCamelCase     ,						dim=0)

										UpperCamelCase__    :		int     =				torch.cat([negative_image_embeds, image_embeds]     ,						dim=0).to(dtype=self.unet.dtype     ,						device=UpperCamelCase)
										UpperCamelCase__    :		Tuple     =				torch.cat([hint, hint]     ,						dim=0).to(dtype=self.unet.dtype     ,						device=UpperCamelCase)

										if not isinstance(UpperCamelCase     ,						UpperCamelCase):
													UpperCamelCase__    :		Tuple     =				[image]
										if not all(isinstance(UpperCamelCase     ,						(PIL.Image.Image, torch.Tensor)) for i in image):
													raise ValueError(
													    F"""Input is in incorrect format: {[type(UpperCamelCase) for i in image]}. Currently, we only support  PIL image and pytorch tensor""")

										UpperCamelCase__    :		List[str]     =				torch.cat([prepare_image(UpperCamelCase     ,						UpperCamelCase     ,						UpperCamelCase) for i in image]     ,						dim=0)
										UpperCamelCase__    :		int     =				image.to(dtype=image_embeds.dtype     ,						device=UpperCamelCase)

										UpperCamelCase__    :		List[Any]     =				self.movq.encode(UpperCamelCase)['latents']
										UpperCamelCase__    :		str     =				latents.repeat_interleave(UpperCamelCase     ,						dim=0)
										self.scheduler.set_timesteps(UpperCamelCase     ,						device=UpperCamelCase)
										UpperCamelCase__   ,			UpperCamelCase__    :		List[str]     =				self.get_timesteps(UpperCamelCase     ,						UpperCamelCase     ,						UpperCamelCase)
										UpperCamelCase__    :		List[str]     =				timesteps[:1].repeat(batch_size * num_images_per_prompt)
										UpperCamelCase__   ,			UpperCamelCase__    :		Optional[Any]     =				downscale_height_and_width(UpperCamelCase     ,						UpperCamelCase     ,						self.movq_scale_factor)
										UpperCamelCase__    :		int     =				self.prepare_latents(
										    UpperCamelCase     ,						UpperCamelCase     ,						UpperCamelCase     ,						UpperCamelCase     ,						image_embeds.dtype     ,						UpperCamelCase     ,						UpperCamelCase)
										for i, t in enumerate(self.progress_bar(UpperCamelCase)):
													# expand the latents if we are doing classifier free guidance
													UpperCamelCase__    :		List[Any]     =				torch.cat([latents] * 2) if do_classifier_free_guidance else latents

													UpperCamelCase__    :		List[str]     =				{'image_embeds': image_embeds, 'hint': hint}
													UpperCamelCase__    :		List[Any]     =				self.unet(
													    sample=UpperCamelCase     ,						timestep=UpperCamelCase     ,						encoder_hidden_states=UpperCamelCase     ,						added_cond_kwargs=UpperCamelCase     ,						return_dict=UpperCamelCase     ,						)[0]

													if do_classifier_free_guidance:
																UpperCamelCase__   ,			UpperCamelCase__    :		str     =				noise_pred.split(latents.shape[1]     ,						dim=1)
																UpperCamelCase__   ,			UpperCamelCase__    :		Tuple     =				noise_pred.chunk(2)
																UpperCamelCase__   ,			UpperCamelCase__    :		Tuple     =				variance_pred.chunk(2)
																UpperCamelCase__    :		Tuple     =				noise_pred_uncond + guidance_scale * (noise_pred_text - noise_pred_uncond)
																UpperCamelCase__    :		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"]
													):
																UpperCamelCase__   ,			UpperCamelCase__    :		str     =				noise_pred.split(latents.shape[1]     ,						dim=1)

													# compute the previous noisy sample x_t -> x_t-1

													UpperCamelCase__    :		Union[str, Any]     =				self.scheduler.step(
													    UpperCamelCase     ,						UpperCamelCase     ,						UpperCamelCase     ,						generator=UpperCamelCase     ,						)[0]

										# post-processing
										UpperCamelCase__    :		Optional[Any]     =				self.movq.decode(UpperCamelCase     ,						force_not_quantize=UpperCamelCase)['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"]:
													UpperCamelCase__    :		Any     =				image * 0.5 + 0.5
													UpperCamelCase__    :		Tuple     =				image.clamp(0     ,						1)
													UpperCamelCase__    :		Tuple     =				image.cpu().permute(0     ,						2     ,						3     ,						1).float().numpy()

										if output_type == "pil":
													UpperCamelCase__    :		Dict     =				self.numpy_to_pil(UpperCamelCase)

										if not return_dict:
													return (image,)

										return ImagePipelineOutput(images=UpperCamelCase)



