






"""simple docstring"""
from typing import List, Optional, Union

import torch

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,
)


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

_lowerCAmelCase         =							"\n    Examples:\n        ```py\n        >>> from diffusers import KandinskyV22Pipeline, KandinskyV22PriorPipeline\n        >>> import torch\n\n        >>> pipe_prior = KandinskyV22PriorPipeline.from_pretrained(\"kandinsky-community/kandinsky-2-2-prior\")\n        >>> pipe_prior.to(\"cuda\")\n        >>> prompt = \"red cat, 4k photo\"\n        >>> out = pipe_prior(prompt)\n        >>> image_emb = out.image_embeds\n        >>> zero_image_emb = out.negative_image_embeds\n        >>> pipe = KandinskyV22Pipeline.from_pretrained(\"kandinsky-community/kandinsky-2-2-decoder\")\n        >>> pipe.to(\"cuda\")\n        >>> image = pipe(\n        ...     image_embeds=image_emb,\n        ...     negative_image_embeds=zero_image_emb,\n        ...     height=768,\n        ...     width=768,\n        ...     num_inference_steps=50,\n        ... ).images\n        >>> image[0].save(\"cat.png\")\n        ```\n"






def 			__UpperCamelCase	(		snake_case__   ,   snake_case__   ,   snake_case__=8					):
	A_		:						int	    =   height // scale_factor**2
	if height % scale_factor**2 != 0:
		new_height += 1
	A_		:						Optional[int]	    =   width // scale_factor**2
	if width % scale_factor**2 != 0:
		new_width += 1
	return new_height * scale_factor, new_width * scale_factor




class      SCREAMING_SNAKE_CASE  (				_SCREAMING_SNAKE_CASE						):





	"""simple docstring"""




	def __init__(self     ,     lowerCAmelCase_     ,     lowerCAmelCase_     ,     lowerCAmelCase_     ,     ):
		super().__init__()

		self.register_modules(
		    unet=lowerCAmelCase_     ,     scheduler=lowerCAmelCase_     ,     movq=lowerCAmelCase_     ,     )
		A_		:						Dict	    =   2 ** (len(self.movq.config.block_out_channels							) - 1)





	def    lowerCamelCase(self     ,     lowerCAmelCase_     ,     lowerCAmelCase_     ,     lowerCAmelCase_     ,     lowerCAmelCase_     ,     lowerCAmelCase_     ,     lowerCAmelCase_							):
		if latents is None:
			A_		:						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}"""							)
			A_		:						Optional[int]	    =   latents.to(lowerCAmelCase_							)

		A_		:						List[str]	    =   latents * scheduler.init_noise_sigma
		return latents





	def    lowerCamelCase(self     ,     lowerCAmelCase_=0							):
		if is_accelerate_available():
			from accelerate import cpu_offload
		else:
			raise ImportError("""Please install accelerate via `pip install accelerate`"""							)

		A_		:						Dict	    =   torch.device(f"""cuda:{gpu_id}"""							)

		A_		:						Any	    =   [
		    self.unet,
		    self.movq,
		]
		for cpu_offloaded_model in models:
			if cpu_offloaded_model is not None:
				cpu_offload(lowerCAmelCase_     ,     lowerCAmelCase_							)





	def    lowerCamelCase(self     ,     lowerCAmelCase_=0							):
		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."""							)

		A_		:						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)

		A_		:						List[str]	    =   None
		for cpu_offloaded_model in [self.unet, self.movq]:
			A_    ,     A_		:						List[str]	    =   cpu_offload_with_hook(lowerCAmelCase_     ,     lowerCAmelCase_     ,     prev_module_hook=lowerCAmelCase_							)

		# We'll offload the last model manually.
		A_		:						Union[str, Any]	    =   hook





	@property
	# Copied from diffusers.pipelines.stable_diffusion.pipeline_stable_diffusion.StableDiffusionPipeline._execution_device
	def    lowerCamelCase(self							):
		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     ,     lowerCAmelCase_     ,     lowerCAmelCase_     ,     lowerCAmelCase_ = 512     ,     lowerCAmelCase_ = 512     ,     lowerCAmelCase_ = 100     ,     lowerCAmelCase_ = 4.0     ,     lowerCAmelCase_ = 1     ,     lowerCAmelCase_ = None     ,     lowerCAmelCase_ = None     ,     lowerCAmelCase_ = "pil"     ,     lowerCAmelCase_ = True     ,     ):
		A_		:						List[Any]	    =   self._execution_device

		A_		:						str	    =   guidance_scale > 1.0

		if isinstance(lowerCAmelCase_     ,     lowerCAmelCase_							):
			A_		:						Dict	    =   torch.cat(lowerCAmelCase_     ,     dim=0							)
		A_		:						Any	    =   image_embeds.shape[0] * num_images_per_prompt
		if isinstance(lowerCAmelCase_     ,     lowerCAmelCase_							):
			A_		:						Optional[int]	    =   torch.cat(lowerCAmelCase_     ,     dim=0							)

		if do_classifier_free_guidance:
			A_		:						Optional[Any]	    =   image_embeds.repeat_interleave(lowerCAmelCase_     ,     dim=0							)
			A_		:						List[str]	    =   negative_image_embeds.repeat_interleave(lowerCAmelCase_     ,     dim=0							)

		A_		:						int	    =   torch.cat([negative_image_embeds, image_embeds]     ,     dim=0							).to(dtype=self.unet.dtype     ,     device=lowerCAmelCase_							)

		self.scheduler.set_timesteps(lowerCAmelCase_     ,     device=lowerCAmelCase_							)
		A_		:						Optional[int]	    =   self.scheduler.timesteps

		A_		:						str	    =   self.unet.config.in_channels

		A_    ,     A_		:						List[Any]	    =   downscale_height_and_width(lowerCAmelCase_     ,     lowerCAmelCase_     ,     self.movq_scale_factor							)

		# create initial latent
		A_		:						Optional[int]	    =   self.prepare_latents(
		    (batch_size, num_channels_latents, height, width)     ,     image_embeds.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
			A_		:						str	    =   torch.cat([latents] * 2							) if do_classifier_free_guidance else latents

			A_		:						Dict	    =   {"""image_embeds""": image_embeds}
			A_		:						Dict	    =   self.unet(
			    sample=lowerCAmelCase_     ,     timestep=lowerCAmelCase_     ,     encoder_hidden_states=lowerCAmelCase_     ,     added_cond_kwargs=lowerCAmelCase_     ,     return_dict=lowerCAmelCase_     ,     )[0]

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

			# compute the previous noisy sample x_t -> x_t-1
			A_		:						int	    =   self.scheduler.step(
			    lowerCAmelCase_     ,     lowerCAmelCase_     ,     lowerCAmelCase_     ,     generator=lowerCAmelCase_     ,     )[0]
		# post-processing
		A_		:						Any	    =   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"]:
			A_		:						Optional[int]	    =   image * 0.5 + 0.5
			A_		:						Dict	    =   image.clamp(0     ,     1							)
			A_		:						Tuple	    =   image.cpu().permute(0     ,     2     ,     3     ,     1							).float().numpy()

		if output_type == "pil":
			A_		:						Optional[Any]	    =   self.numpy_to_pil(lowerCAmelCase_							)

		if not return_dict:
			return (image,)

		return ImagePipelineOutput(images=lowerCAmelCase_							)









"""simple docstring"""
# Copyright 2023 The HuggingFace Inc. team. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import torch

from ..models.auto import AutoModelForSequenceClassification, AutoTokenizer
from .base import PipelineTool



class      SCREAMING_SNAKE_CASE  (				_SCREAMING_SNAKE_CASE						):





	"""simple docstring"""




	_A       :				Optional[int]    =				"""facebook/bart-large-mnli"""
	_A       :				str    =				(
	    """This is a tool that classifies an English text using provided labels. It takes two inputs: `text`, which """
	    """should be the text to classify, and `labels`, which should be the list of labels to use for classification. """
	    """It returns the most likely label in the list of provided `labels` for the input text."""
	)
	_A       :				List[str]    =				"""text_classifier"""
	_A       :				Optional[int]    =				AutoTokenizer
	_A       :				Optional[Any]    =				AutoModelForSequenceClassification

	_A       :				List[str]    =				["""text""", ["""text"""]]
	_A       :				Dict    =				["""text"""]





	def    lowerCamelCase(self							):
		super().setup()
		A_		:						int	    =   self.model.config
		A_		:						List[str]	    =   -1
		for idx, label in config.idalabel.items():
			if label.lower().startswith("""entail"""							):
				A_		:						List[Any]	    =   int(lowerCAmelCase_							)
		if self.entailment_id == -1:
			raise ValueError("""Could not determine the entailment ID from the model config, please pass it at init."""							)





	def    lowerCamelCase(self     ,     lowerCAmelCase_     ,     lowerCAmelCase_							):
		A_		:						List[Any]	    =   labels
		return self.pre_processor(
		    [text] * len(lowerCAmelCase_							)     ,     [f"""This example is {label}""" for label in labels]     ,     return_tensors="""pt"""     ,     padding="""max_length"""     ,     )





	def    lowerCamelCase(self     ,     lowerCAmelCase_							):
		A_		:						str	    =   outputs.logits
		A_		:						Optional[int]	    =   torch.argmax(logits[:, 2]							).item()
		return self._labels[label_id]

