






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

import torch
from transformers import CLIPImageProcessor, CLIPTextModel, CLIPTokenizer

from diffusers import DiffusionPipeline
from diffusers.models import AutoencoderKL, UNetaDConditionModel
from diffusers.pipelines.stable_diffusion import StableDiffusionPipelineOutput
from diffusers.pipelines.stable_diffusion.safety_checker import StableDiffusionSafetyChecker
from diffusers.schedulers import DDIMScheduler, LMSDiscreteScheduler, PNDMScheduler
from diffusers.utils import logging


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



class      SCREAMING_SNAKE_CASE  (				__a						):





	"""simple docstring"""




	def __init__(self     ,     lowerCAmelCase_     ,     lowerCAmelCase_     ,     lowerCAmelCase_     ,     lowerCAmelCase_     ,     lowerCAmelCase_     ,     lowerCAmelCase_     ,     lowerCAmelCase_     ,     ):
		super().__init__()
		self.register_modules(
		    vae=A__     ,     text_encoder=A__     ,     tokenizer=A__     ,     unet=A__     ,     scheduler=A__     ,     safety_checker=A__     ,     feature_extractor=A__     ,     )





	def    lowerCamelCase(self     ,     lowerCAmelCase_ = "auto"							):
		if slice_size == "auto":
			# half the attention head size is usually a good trade-off between
			# speed and memory
			A_		:						str	    =   self.unet.config.attention_head_dim // 2
		self.unet.set_attention_slice(A__							)





	def    lowerCamelCase(self							):
		self.enable_attention_slicing(A__							)





	@torch.no_grad()
	def __call__(self     ,     lowerCAmelCase_     ,     lowerCAmelCase_ = 512     ,     lowerCAmelCase_ = 512     ,     lowerCAmelCase_ = 50     ,     lowerCAmelCase_ = 7.5     ,     lowerCAmelCase_ = None     ,     lowerCAmelCase_ = 1     ,     lowerCAmelCase_ = 0.0     ,     lowerCAmelCase_ = None     ,     lowerCAmelCase_ = None     ,     lowerCAmelCase_ = "pil"     ,     lowerCAmelCase_ = True     ,     lowerCAmelCase_ = None     ,     lowerCAmelCase_ = 1     ,     lowerCAmelCase_ = None     ,     **lowerCAmelCase_     ,     ):

		if isinstance(A__     ,     A__							):
			A_		:						Optional[Any]	    =   1
		elif isinstance(A__     ,     A__							):
			A_		:						Union[str, Any]	    =   len(A__							)
		else:
			raise ValueError(f"""`prompt` has to be of type `str` or `list` but is {type(A__							)}"""							)

		if height % 8 != 0 or width % 8 != 0:
			raise ValueError(f"""`height` and `width` have to be divisible by 8 but are {height} and {width}."""							)

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

		# get prompt text embeddings
		A_		:						List[str]	    =   self.tokenizer(
		    A__     ,     padding="""max_length"""     ,     max_length=self.tokenizer.model_max_length     ,     return_tensors="""pt"""     ,     )
		A_		:						Any	    =   text_inputs.input_ids

		if text_input_ids.shape[-1] > self.tokenizer.model_max_length:
			A_		:						Optional[int]	    =   self.tokenizer.batch_decode(text_input_ids[:, self.tokenizer.model_max_length :]							)
			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}"""							)
			A_		:						Optional[int]	    =   text_input_ids[:, : self.tokenizer.model_max_length]

		if text_embeddings is None:
			A_		:						Optional[Any]	    =   self.text_encoder(text_input_ids.to(self.device							)							)[0]

		# duplicate text embeddings for each generation per prompt, using mps friendly method
		A_    ,     A_    ,     A_		:						Union[str, Any]	    =   text_embeddings.shape
		A_		:						Union[str, Any]	    =   text_embeddings.repeat(1     ,     A__     ,     1							)
		A_		:						int	    =   text_embeddings.view(bs_embed * num_images_per_prompt     ,     A__     ,     -1							)

		# 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.
		A_		:						int	    =   guidance_scale > 1.0
		# get unconditional embeddings for classifier free guidance
		if do_classifier_free_guidance:
			A_		:						Any	    =   42
			if negative_prompt is None:
				A_		:						str	    =   [""""""]
			elif type(A__							) is not type(A__							):
				raise TypeError(
				    f"""`negative_prompt` should be the same type to `prompt`, but got {type(A__							)} !="""
				    f""" {type(A__							)}."""							)
			elif isinstance(A__     ,     A__							):
				A_		:						str	    =   [negative_prompt]
			elif batch_size != len(A__							):
				raise ValueError(
				    f"""`negative_prompt`: {negative_prompt} has batch size {len(A__							)}, but `prompt`:"""
				    f""" {prompt} has batch size {batch_size}. Please make sure that passed `negative_prompt` matches"""
				    """ the batch size of `prompt`."""							)
			else:
				A_		:						Any	    =   negative_prompt

			A_		:						Optional[Any]	    =   text_input_ids.shape[-1]
			A_		:						int	    =   self.tokenizer(
			    A__     ,     padding="""max_length"""     ,     max_length=A__     ,     truncation=A__     ,     return_tensors="""pt"""     ,     )
			A_		:						Tuple	    =   self.text_encoder(uncond_input.input_ids.to(self.device							)							)[0]

			# duplicate unconditional embeddings for each generation per prompt, using mps friendly method
			A_		:						str	    =   uncond_embeddings.shape[1]
			A_		:						Optional[int]	    =   uncond_embeddings.repeat(A__     ,     A__     ,     1							)
			A_		:						int	    =   uncond_embeddings.view(batch_size * num_images_per_prompt     ,     A__     ,     -1							)

			# 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
			A_		:						Optional[Any]	    =   torch.cat([uncond_embeddings, text_embeddings]							)

		# get the initial random noise unless the user supplied it

		# Unlike in other pipelines, latents need to be generated in the target device
		# for 1-to-1 results reproducibility with the CompVis implementation.
		# However this currently doesn't work in `mps`.
		A_		:						Optional[int]	    =   (batch_size * num_images_per_prompt, self.unet.config.in_channels, height // 8, width // 8)
		A_		:						Any	    =   (batch_size * num_images_per_prompt, self.unet.config.in_channels, 64, 64)
		A_		:						List[Any]	    =   text_embeddings.dtype
		if latents is None:
			if self.device.type == "mps":
				# randn does not exist on mps
				A_		:						List[Any]	    =   torch.randn(
				    A__     ,     generator=A__     ,     device="""cpu"""     ,     dtype=A__							).to(self.device							)
				A_		:						str	    =   torch.randn(A__     ,     generator=A__     ,     device="""cpu"""     ,     dtype=A__							).to(
				    self.device							)
			else:
				A_		:						Optional[Any]	    =   torch.randn(
				    A__     ,     generator=A__     ,     device=self.device     ,     dtype=A__							)
				A_		:						List[Any]	    =   torch.randn(A__     ,     generator=A__     ,     device=self.device     ,     dtype=A__							)
		else:
			if latents_reference.shape != latents_shape:
				raise ValueError(f"""Unexpected latents shape, got {latents.shape}, expected {latents_shape}"""							)
			A_		:						Dict	    =   latents_reference.to(self.device							)
			A_		:						str	    =   latents.to(self.device							)

		# This is the key part of the pipeline where we
		# try to ensure that the generated images w/ the same seed
		# but different sizes actually result in similar images
		A_		:						str	    =   (latents_shape[3] - latents_shape_reference[3]) // 2
		A_		:						Dict	    =   (latents_shape[2] - latents_shape_reference[2]) // 2
		A_		:						Optional[Any]	    =   latents_shape_reference[3] if dx >= 0 else latents_shape_reference[3] + 2 * dx
		A_		:						Union[str, Any]	    =   latents_shape_reference[2] if dy >= 0 else latents_shape_reference[2] + 2 * dy
		A_		:						Any	    =   0 if dx < 0 else dx
		A_		:						Any	    =   0 if dy < 0 else dy
		A_		:						List[Any]	    =   max(-dx     ,     0							)
		A_		:						List[str]	    =   max(-dy     ,     0							)
		# import pdb
		# pdb.set_trace()
		A_		:						Dict	    =   latents_reference[:, :, dy : dy + h, dx : dx + w]

		# set timesteps
		self.scheduler.set_timesteps(A__							)

		# Some schedulers like PNDM have timesteps as arrays
		# It's more optimized to move all timesteps to correct device beforehand
		A_		:						Optional[Any]	    =   self.scheduler.timesteps.to(self.device							)

		# scale the initial noise by the standard deviation required by the scheduler
		A_		:						Dict	    =   latents * self.scheduler.init_noise_sigma

		# 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]
		A_		:						Union[str, Any]	    =   """eta""" in set(inspect.signature(self.scheduler.step							).parameters.keys()							)
		A_		:						str	    =   {}
		if accepts_eta:
			A_		:						Optional[Any]	    =   eta

		for i, t in enumerate(self.progress_bar(A__							)							):
			# expand the latents if we are doing classifier free guidance
			A_		:						List[str]	    =   torch.cat([latents] * 2							) if do_classifier_free_guidance else latents
			A_		:						List[str]	    =   self.scheduler.scale_model_input(A__     ,     A__							)

			# predict the noise residual
			A_		:						Union[str, Any]	    =   self.unet(A__     ,     A__     ,     encoder_hidden_states=A__							).sample

			# perform guidance
			if do_classifier_free_guidance:
				A_    ,     A_		:						List[Any]	    =   noise_pred.chunk(2							)
				A_		:						str	    =   noise_pred_uncond + guidance_scale * (noise_pred_text - noise_pred_uncond)

			# compute the previous noisy sample x_t -> x_t-1
			A_		:						List[str]	    =   self.scheduler.step(A__     ,     A__     ,     A__     ,     **A__							).prev_sample

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

		A_		:						Any	    =   1 / 0.18215 * latents
		A_		:						Optional[Any]	    =   self.vae.decode(A__							).sample

		A_		:						Any	    =   (image / 2 + 0.5).clamp(0     ,     1							)

		# we always cast to float32 as this does not cause significant overhead and is compatible with bfloat16
		A_		:						Optional[Any]	    =   image.cpu().permute(0     ,     2     ,     3     ,     1							).float().numpy()

		if self.safety_checker is not None:
			A_		:						Optional[int]	    =   self.feature_extractor(self.numpy_to_pil(A__							)     ,     return_tensors="""pt"""							).to(
			    self.device							)
			A_    ,     A_		:						int	    =   self.safety_checker(
			    images=A__     ,     clip_input=safety_checker_input.pixel_values.to(text_embeddings.dtype							)							)
		else:
			A_		:						List[str]	    =   None

		if output_type == "pil":
			A_		:						Any	    =   self.numpy_to_pil(A__							)

		if not return_dict:
			return (image, has_nsfw_concept)

		return StableDiffusionPipelineOutput(images=A__     ,     nsfw_content_detected=A__							)









"""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]

