diffuse-custom

Runtime error

App Files Files Community

diffuse-custom / diffusers /pipelines /stable_diffusion /pipeline_onnx_stable_diffusion_inpaint_legacy.py

Jackflack09

Duplicate from YeOldHermit/Super-Resolution-Anime-Diffusion

522606a about 1 year ago

raw history blame contribute delete

No virus

23.7 kB

	import inspect
	from typing import Callable, List, Optional, Union

	import numpy as np
	import torch

	import PIL
	from transformers import CLIPFeatureExtractor, CLIPTokenizer

	from ...configuration_utils import FrozenDict
	from ...onnx_utils import OnnxRuntimeModel
	from ...pipeline_utils import DiffusionPipeline
	from ...schedulers import DDIMScheduler, LMSDiscreteScheduler, PNDMScheduler
	from ...utils import deprecate, logging
	from . import StableDiffusionPipelineOutput


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


	def preprocess(image):
	w, h = image.size
	w, h = map(lambda x: x - x % 32, (w, h)) # resize to integer multiple of 32
	image = image.resize((w, h), resample=PIL.Image.LANCZOS)
	image = np.array(image).astype(np.float32) / 255.0
	image = image[None].transpose(0, 3, 1, 2)
	return 2.0 * image - 1.0


	def preprocess_mask(mask, scale_factor=8):
	mask = mask.convert("L")
	w, h = mask.size
	w, h = map(lambda x: x - x % 32, (w, h)) # resize to integer multiple of 32
	mask = mask.resize((w // scale_factor, h // scale_factor), resample=PIL.Image.NEAREST)
	mask = np.array(mask).astype(np.float32) / 255.0
	mask = np.tile(mask, (4, 1, 1))
	mask = mask[None].transpose(0, 1, 2, 3) # what does this step do?
	mask = 1 - mask # repaint white, keep black
	return mask


	class OnnxStableDiffusionInpaintPipelineLegacy(DiffusionPipeline):
	r"""
	Pipeline for text-guided image inpainting using Stable Diffusion. This is a legacy feature for Onnx pipelines to
	provide compatibility with StableDiffusionInpaintPipelineLegacy and may be removed in the future.

	This model inherits from [`DiffusionPipeline`]. Check the superclass documentation for the generic methods the
	library implements for all the pipelines (such as downloading or saving, running on a particular device, etc.)

	Args:
	vae ([`AutoencoderKL`]):
	Variational Auto-Encoder (VAE) Model to encode and decode images to and from latent representations.
	text_encoder ([`CLIPTextModel`]):
	Frozen text-encoder. Stable Diffusion uses the text portion of
	[CLIP](https://huggingface.co/docs/transformers/model_doc/clip#transformers.CLIPTextModel), specifically
	the [clip-vit-large-patch14](https://huggingface.co/openai/clip-vit-large-patch14) variant.
	tokenizer (`CLIPTokenizer`):
	Tokenizer of class
	[CLIPTokenizer](https://huggingface.co/docs/transformers/v4.21.0/en/model_doc/clip#transformers.CLIPTokenizer).
	unet ([`UNet2DConditionModel`]): Conditional U-Net architecture to denoise the encoded image latents.
	scheduler ([`SchedulerMixin`]):
	A scheduler to be used in combination with `unet` to denoise the encoded image latents. Can be one of
	[`DDIMScheduler`], [`LMSDiscreteScheduler`], or [`PNDMScheduler`].
	safety_checker ([`StableDiffusionSafetyChecker`]):
	Classification module that estimates whether generated images could be considered offensive or harmful.
	Please, refer to the [model card](https://huggingface.co/runwayml/stable-diffusion-v1-5) for details.
	feature_extractor ([`CLIPFeatureExtractor`]):
	Model that extracts features from generated images to be used as inputs for the `safety_checker`.
	"""
	_optional_components = ["safety_checker", "feature_extractor"]

	vae_encoder: OnnxRuntimeModel
	vae_decoder: OnnxRuntimeModel
	text_encoder: OnnxRuntimeModel
	tokenizer: CLIPTokenizer
	unet: OnnxRuntimeModel
	scheduler: Union[DDIMScheduler, PNDMScheduler, LMSDiscreteScheduler]
	safety_checker: OnnxRuntimeModel
	feature_extractor: CLIPFeatureExtractor

	def __init__(
	self,
	vae_encoder: OnnxRuntimeModel,
	vae_decoder: OnnxRuntimeModel,
	text_encoder: OnnxRuntimeModel,
	tokenizer: CLIPTokenizer,
	unet: OnnxRuntimeModel,
	scheduler: Union[DDIMScheduler, PNDMScheduler, LMSDiscreteScheduler],
	safety_checker: OnnxRuntimeModel,
	feature_extractor: CLIPFeatureExtractor,
	requires_safety_checker: bool = True,
	):
	super().__init__()

	if hasattr(scheduler.config, "steps_offset") and scheduler.config.steps_offset != 1:
	deprecation_message = (
	f"The configuration file of this scheduler: {scheduler} is outdated. `steps_offset`"
	f" should be set to 1 instead of {scheduler.config.steps_offset}. Please make sure "
	"to update the config accordingly as leaving `steps_offset` might led to incorrect results"
	" in future versions. If you have downloaded this checkpoint from the Hugging Face Hub,"
	" it would be very nice if you could open a Pull request for the `scheduler/scheduler_config.json`"
	" file"
	)
	deprecate("steps_offset!=1", "1.0.0", deprecation_message, standard_warn=False)
	new_config = dict(scheduler.config)
	new_config["steps_offset"] = 1
	scheduler._internal_dict = FrozenDict(new_config)

	if hasattr(scheduler.config, "clip_sample") and scheduler.config.clip_sample is True:
	deprecation_message = (
	f"The configuration file of this scheduler: {scheduler} has not set the configuration `clip_sample`."
	" `clip_sample` should be set to False in the configuration file. Please make sure to update the"
	" config accordingly as not setting `clip_sample` in the config might lead to incorrect results in"
	" future versions. If you have downloaded this checkpoint from the Hugging Face Hub, it would be very"
	" nice if you could open a Pull request for the `scheduler/scheduler_config.json` file"
	)
	deprecate("clip_sample not set", "1.0.0", deprecation_message, standard_warn=False)
	new_config = dict(scheduler.config)
	new_config["clip_sample"] = False
	scheduler._internal_dict = FrozenDict(new_config)

	if safety_checker is None and requires_safety_checker:
	logger.warning(
	f"You have disabled the safety checker for {self.__class__} by passing `safety_checker=None`. Ensure"
	" that you abide to the conditions of the Stable Diffusion license and do not expose unfiltered"
	" results in services or applications open to the public. Both the diffusers team and Hugging Face"
	" strongly recommend to keep the safety filter enabled in all public facing circumstances, disabling"
	" it only for use-cases that involve analyzing network behavior or auditing its results. For more"
	" information, please have a look at https://github.com/huggingface/diffusers/pull/254 ."
	)

	if safety_checker is not None and feature_extractor is None:
	raise ValueError(
	"Make sure to define a feature extractor when loading {self.__class__} if you want to use the safety"
	" checker. If you do not want to use the safety checker, you can pass `'safety_checker=None'` instead."
	)

	self.register_modules(
	vae_encoder=vae_encoder,
	vae_decoder=vae_decoder,
	text_encoder=text_encoder,
	tokenizer=tokenizer,
	unet=unet,
	scheduler=scheduler,
	safety_checker=safety_checker,
	feature_extractor=feature_extractor,
	)
	self.register_to_config(requires_safety_checker=requires_safety_checker)

	# Copied from diffusers.pipelines.stable_diffusion.pipeline_onnx_stable_diffusion.OnnxStableDiffusionPipeline._encode_prompt
	def _encode_prompt(self, prompt, num_images_per_prompt, do_classifier_free_guidance, negative_prompt):
	r"""
	Encodes the prompt into text encoder hidden states.

	Args:
	prompt (`str` or `list(int)`):
	prompt to be encoded
	num_images_per_prompt (`int`):
	number of images that should be generated per prompt
	do_classifier_free_guidance (`bool`):
	whether to use classifier free guidance or not
	negative_prompt (`str` or `List[str]`):
	The prompt or prompts not to guide the image generation. Ignored when not using guidance (i.e., ignored
	if `guidance_scale` is less than `1`).
	"""
	batch_size = len(prompt) if isinstance(prompt, list) else 1

	# get prompt text embeddings
	text_inputs = self.tokenizer(
	prompt,
	padding="max_length",
	max_length=self.tokenizer.model_max_length,
	truncation=True,
	return_tensors="np",
	)
	text_input_ids = text_inputs.input_ids
	untruncated_ids = self.tokenizer(prompt, padding="max_length", return_tensors="np").input_ids

	if not np.array_equal(text_input_ids, untruncated_ids):
	removed_text = 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}"
	)

	text_embeddings = self.text_encoder(input_ids=text_input_ids.astype(np.int32))[0]
	text_embeddings = np.repeat(text_embeddings, num_images_per_prompt, axis=0)

	# get unconditional embeddings for classifier free guidance
	if do_classifier_free_guidance:
	uncond_tokens: List[str]
	if negative_prompt is None:
	uncond_tokens = [""] * batch_size
	elif type(prompt) is not type(negative_prompt):
	raise TypeError(
	f"`negative_prompt` should be the same type to `prompt`, but got {type(negative_prompt)} !="
	f" {type(prompt)}."
	)
	elif isinstance(negative_prompt, str):
	uncond_tokens = [negative_prompt] * batch_size
	elif batch_size != len(negative_prompt):
	raise ValueError(
	f"`negative_prompt`: {negative_prompt} has batch size {len(negative_prompt)}, but `prompt`:"
	f" {prompt} has batch size {batch_size}. Please make sure that passed `negative_prompt` matches"
	" the batch size of `prompt`."
	)
	else:
	uncond_tokens = negative_prompt

	max_length = text_input_ids.shape[-1]
	uncond_input = self.tokenizer(
	uncond_tokens,
	padding="max_length",
	max_length=max_length,
	truncation=True,
	return_tensors="np",
	)
	uncond_embeddings = self.text_encoder(input_ids=uncond_input.input_ids.astype(np.int32))[0]
	uncond_embeddings = np.repeat(uncond_embeddings, num_images_per_prompt, axis=0)

	# 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
	text_embeddings = np.concatenate([uncond_embeddings, text_embeddings])

	return text_embeddings

	def __call__(
	self,
	prompt: Union[str, List[str]],
	image: Union[np.ndarray, PIL.Image.Image],
	mask_image: Union[np.ndarray, PIL.Image.Image],
	strength: float = 0.8,
	num_inference_steps: Optional[int] = 50,
	guidance_scale: Optional[float] = 7.5,
	negative_prompt: Optional[Union[str, List[str]]] = None,
	num_images_per_prompt: Optional[int] = 1,
	eta: Optional[float] = 0.0,
	generator: Optional[np.random.RandomState] = None,
	output_type: Optional[str] = "pil",
	return_dict: bool = True,
	callback: Optional[Callable[[int, int, np.ndarray], None]] = None,
	callback_steps: Optional[int] = 1,
	**kwargs,
	):
	r"""
	Function invoked when calling the pipeline for generation.

	Args:
	prompt (`str` or `List[str]`):
	The prompt or prompts to guide the image generation.
	image (`nd.ndarray` or `PIL.Image.Image`):
	`Image`, or tensor representing an image batch, that will be used as the starting point for the
	process. This is the image whose masked region will be inpainted.
	mask_image (`nd.ndarray` or `PIL.Image.Image`):
	`Image`, or tensor representing an image batch, to mask `image`. White pixels in the mask will be
	replaced by noise and therefore repainted, while black pixels will be preserved. If `mask_image` is a
	PIL image, it will be converted to a single channel (luminance) before use. If it's a tensor, it should
	contain one color channel (L) instead of 3, so the expected shape would be `(B, H, W, 1)`.uu
	strength (`float`, optional, defaults to 0.8):
	Conceptually, indicates how much to transform the reference `image`. Must be between 0 and 1. `image`
	will be used as a starting point, adding more noise to it the larger the `strength`. The number of
	denoising steps depends on the amount of noise initially added. When `strength` is 1, added noise will
	be maximum and the denoising process will run for the full number of iterations specified in
	`num_inference_steps`. A value of 1, therefore, essentially ignores `image`.
	num_inference_steps (`int`, optional, defaults to 50):
	The number of denoising steps. More denoising steps usually lead to a higher quality image at the
	expense of slower inference. This parameter will be modulated by `strength`.
	guidance_scale (`float`, optional, defaults to 7.5):
	Guidance scale as defined in [Classifier-Free Diffusion Guidance](https://arxiv.org/abs/2207.12598).
	`guidance_scale` is defined as `w` of equation 2. of [Imagen
	Paper](https://arxiv.org/pdf/2205.11487.pdf). Guidance scale is enabled by setting `guidance_scale >
	1`. Higher guidance scale encourages to generate images that are closely linked to the text `prompt`,
	usually at the expense of lower image quality.
	negative_prompt (`str` or `List[str]`, optional):
	The prompt or prompts not to guide the image generation. Ignored when not using guidance (i.e., ignored
	if `guidance_scale` is less than `1`).
	num_images_per_prompt (`int`, optional, defaults to 1):
	The number of images to generate per prompt.
	eta (`float`, optional, defaults to 0.0):
	Corresponds to parameter eta (?) in the DDIM paper: https://arxiv.org/abs/2010.02502. Only applies to
	[`schedulers.DDIMScheduler`], will be ignored for others.
	generator (`np.random.RandomState`, optional):
	A np.random.RandomState to make generation deterministic.
	output_type (`str`, optional, defaults to `"pil"`):
	The output format of the generate image. Choose between
	[PIL](https://pillow.readthedocs.io/en/stable/): `PIL.Image.Image` or `np.array`.
	return_dict (`bool`, optional, defaults to `True`):
	Whether or not to return a [`~pipelines.stable_diffusion.StableDiffusionPipelineOutput`] instead of a
	plain tuple.
	callback (`Callable`, optional):
	A function that will be called every `callback_steps` steps during inference. The function will be
	called with the following arguments: `callback(step: int, timestep: int, latents: np.ndarray)`.
	callback_steps (`int`, optional, defaults to 1):
	The frequency at which the `callback` function will be called. If not specified, the callback will be
	called at every step.

	Returns:
	[`~pipelines.stable_diffusion.StableDiffusionPipelineOutput`] or `tuple`:
	[`~pipelines.stable_diffusion.StableDiffusionPipelineOutput`] if `return_dict` is True, otherwise a `tuple.
	When returning a tuple, the first element is a list with the generated images, and the second element is a
	list of `bool`s denoting whether the corresponding generated image likely represents "not-safe-for-work"
	(nsfw) content, according to the `safety_checker`.
	"""
	message = "Please use `image` instead of `init_image`."
	init_image = deprecate("init_image", "0.12.0", message, take_from=kwargs)
	image = init_image or image

	if isinstance(prompt, str):
	batch_size = 1
	elif isinstance(prompt, list):
	batch_size = len(prompt)
	else:
	raise ValueError(f"`prompt` has to be of type `str` or `list` but is {type(prompt)}")

	if strength < 0 or strength > 1:
	raise ValueError(f"The value of strength should in [0.0, 1.0] but is {strength}")

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

	if generator is None:
	generator = np.random

	# set timesteps
	self.scheduler.set_timesteps(num_inference_steps)

	if isinstance(image, PIL.Image.Image):
	image = preprocess(image)

	# here `guidance_scale` is defined analog to the guidance weight `w` of equation (2)
	# of the Imagen paper: https://arxiv.org/pdf/2205.11487.pdf . `guidance_scale = 1`
	# corresponds to doing no classifier free guidance.
	do_classifier_free_guidance = guidance_scale > 1.0

	text_embeddings = self._encode_prompt(
	prompt, num_images_per_prompt, do_classifier_free_guidance, negative_prompt
	)

	latents_dtype = text_embeddings.dtype
	image = image.astype(latents_dtype)

	# encode the init image into latents and scale the latents
	init_latents = self.vae_encoder(sample=image)[0]
	init_latents = 0.18215 * init_latents

	# Expand init_latents for batch_size and num_images_per_prompt
	init_latents = np.concatenate([init_latents] * num_images_per_prompt, axis=0)
	init_latents_orig = init_latents

	# preprocess mask
	if not isinstance(mask_image, np.ndarray):
	mask_image = preprocess_mask(mask_image, 8)
	mask_image = mask_image.astype(latents_dtype)
	mask = np.concatenate([mask_image] * num_images_per_prompt, axis=0)

	# check sizes
	if not mask.shape == init_latents.shape:
	raise ValueError("The mask and image should be the same size!")

	# get the original timestep using init_timestep
	offset = self.scheduler.config.get("steps_offset", 0)
	init_timestep = int(num_inference_steps * strength) + offset
	init_timestep = min(init_timestep, num_inference_steps)

	timesteps = self.scheduler.timesteps.numpy()[-init_timestep]
	timesteps = np.array([timesteps] * batch_size * num_images_per_prompt)

	# add noise to latents using the timesteps
	noise = generator.randn(*init_latents.shape).astype(latents_dtype)
	init_latents = self.scheduler.add_noise(
	torch.from_numpy(init_latents), torch.from_numpy(noise), torch.from_numpy(timesteps)
	)
	init_latents = init_latents.numpy()

	# prepare extra kwargs for the scheduler step, since not all schedulers have the same signature
	# eta (?) is only used with the DDIMScheduler, it will be ignored for other schedulers.
	# eta corresponds to ? in DDIM paper: https://arxiv.org/abs/2010.02502
	# and should be between [0, 1]
	accepts_eta = "eta" in set(inspect.signature(self.scheduler.step).parameters.keys())
	extra_step_kwargs = {}
	if accepts_eta:
	extra_step_kwargs["eta"] = eta

	latents = init_latents

	t_start = max(num_inference_steps - init_timestep + offset, 0)
	timesteps = self.scheduler.timesteps[t_start:].numpy()

	for i, t in enumerate(self.progress_bar(timesteps)):
	# expand the latents if we are doing classifier free guidance
	latent_model_input = np.concatenate([latents] * 2) if do_classifier_free_guidance else latents
	latent_model_input = self.scheduler.scale_model_input(latent_model_input, t)

	# predict the noise residual
	noise_pred = self.unet(
	sample=latent_model_input, timestep=np.array([t]), encoder_hidden_states=text_embeddings
	)[0]

	# perform guidance
	if do_classifier_free_guidance:
	noise_pred_uncond, noise_pred_text = np.split(noise_pred, 2)
	noise_pred = noise_pred_uncond + guidance_scale * (noise_pred_text - noise_pred_uncond)

	# compute the previous noisy sample x_t -> x_t-1
	latents = self.scheduler.step(
	torch.from_numpy(noise_pred), t, torch.from_numpy(latents), **extra_step_kwargs
	).prev_sample

	latents = latents.numpy()

	init_latents_proper = self.scheduler.add_noise(
	torch.from_numpy(init_latents_orig), torch.from_numpy(noise), torch.from_numpy(np.array([t]))
	)

	init_latents_proper = init_latents_proper.numpy()

	latents = (init_latents_proper * mask) + (latents * (1 - mask))

	# call the callback, if provided
	if callback is not None and i % callback_steps == 0:
	callback(i, t, latents)

	latents = 1 / 0.18215 * latents
	# image = self.vae_decoder(latent_sample=latents)[0]
	# it seems likes there is a strange result for using half-precision vae decoder if batchsize>1
	image = np.concatenate(
	[self.vae_decoder(latent_sample=latents[i : i + 1])[0] for i in range(latents.shape[0])]
	)

	image = np.clip(image / 2 + 0.5, 0, 1)
	image = image.transpose((0, 2, 3, 1))

	if self.safety_checker is not None:
	safety_checker_input = self.feature_extractor(
	self.numpy_to_pil(image), return_tensors="np"
	).pixel_values.astype(image.dtype)
	# There will throw an error if use safety_checker batchsize>1
	images, has_nsfw_concept = [], []
	for i in range(image.shape[0]):
	image_i, has_nsfw_concept_i = self.safety_checker(
	clip_input=safety_checker_input[i : i + 1], images=image[i : i + 1]
	)
	images.append(image_i)
	has_nsfw_concept.append(has_nsfw_concept_i[0])
	image = np.concatenate(images)
	else:
	has_nsfw_concept = None

	if output_type == "pil":
	image = self.numpy_to_pil(image)

	if not return_dict:
	return (image, has_nsfw_concept)

	return StableDiffusionPipelineOutput(images=image, nsfw_content_detected=has_nsfw_concept)