




import gzip
import hashlib
import json
import multiprocessing
import os
import re
import shutil
import time
from pathlib import Path

import numpy as np
from arguments import PreprocessingArguments
from datasets import load_dataset
from minhash_deduplication import deduplicate_dataset

from transformers import AutoTokenizer, HfArgumentParser


lowerCamelCase	:	List[Any]				=   re.compile(r'\s+')




def 						SCREAMING_SNAKE_CASE__ (   lowercase					)  ->       List[str]:
   return {"hash": hashlib.mda(re.sub(lowercase						,""""""						,example["""content"""]					).encode("""utf-8"""					)					).hexdigest()}




def 						SCREAMING_SNAKE_CASE__ (   lowercase					)  ->       Optional[Any]:
   snake_case				:    Union[str, Any]           =	[len(lowercase					) for line in example["""content"""].splitlines()]
   return {"line_mean": np.mean(lowercase					), "line_max": max(lowercase					)}




def 						SCREAMING_SNAKE_CASE__ (   lowercase					)  ->       Optional[Any]:
   snake_case				:    str           =	np.mean([c.isalnum() for c in example["""content"""]]					)
   return {"alpha_frac": alpha_frac}




def 						SCREAMING_SNAKE_CASE__ (   lowercase						,lowercase					)  ->       Optional[Any]:
   if example["hash"] in uniques:
      uniques.remove(example["""hash"""]					)
      return True
   else:
      return False




def 						SCREAMING_SNAKE_CASE__ (   lowercase						,lowercase=5					)  ->       List[str]:
   snake_case				:    Tuple           =	["""auto-generated""", """autogenerated""", """automatically generated"""]
   snake_case				:    str           =	example["""content"""].splitlines()
   for _, line in zip(range(lowercase					)						,lowercase					):
      for keyword in keywords:
         if keyword in line.lower():
            return {"autogenerated": True}
   else:
      return {"autogenerated": False}




def 						SCREAMING_SNAKE_CASE__ (   lowercase						,lowercase=5						,lowercase=0.05					)  ->       Any:
   snake_case				:    Union[str, Any]           =	["""unit tests""", """test file""", """configuration file"""]
   snake_case				:    int           =	example["""content"""].splitlines()
   snake_case				:    int           =	0
   snake_case				:    List[Any]           =	0
   # first test
   for _, line in zip(range(lowercase					)						,lowercase					):
      for keyword in keywords:
         if keyword in line.lower():
            return {"config_or_test": True}
    # second test
   snake_case				:    str           =	example["""content"""].count("""\n"""					)
   snake_case				:    int           =	int(coeff * nlines					)
   for line in lines:
      count_config += line.lower().count("""config"""					)
      count_test += line.lower().count("""test"""					)
      if count_config > threshold or count_test > threshold:
         return {"config_or_test": True}
   return {"config_or_test": False}




def 						SCREAMING_SNAKE_CASE__ (   lowercase					)  ->       Optional[int]:
   snake_case				:    Any           =	["""def """, """class """, """for """, """while """]
   snake_case				:    Tuple           =	example["""content"""].splitlines()
   for line in lines:
      for keyword in keywords:
         if keyword in line.lower():
            return {"has_no_keywords": False}
   return {"has_no_keywords": True}




def 						SCREAMING_SNAKE_CASE__ (   lowercase						,lowercase=4					)  ->       Any:
   snake_case				:    List[str]           =	example["""content"""].splitlines()
   snake_case				:    List[Any]           =	0
   for line in lines:
      counter += line.lower().count("""="""					)
      if counter > minimum:
         return {"has_few_assignments": False}
   return {"has_few_assignments": True}




def 						SCREAMING_SNAKE_CASE__ (   lowercase					)  ->       Optional[int]:
   snake_case				:    List[Any]           =	tokenizer(example["""content"""]						,truncation=lowercase					)["""input_ids"""]
   snake_case				:    str           =	len(example["""content"""]					) / len(lowercase					)
   return {"ratio": ratio}




def 						SCREAMING_SNAKE_CASE__ (   lowercase					)  ->       List[str]:
   snake_case				:    Optional[Any]           =	{}
   results.update(get_hash(lowercase					)					)
   results.update(line_stats(lowercase					)					)
   results.update(alpha_stats(lowercase					)					)
   results.update(char_token_ratio(lowercase					)					)
   results.update(is_autogenerated(lowercase					)					)
   results.update(is_config_or_test(lowercase					)					)
   results.update(has_no_keywords(lowercase					)					)
   results.update(has_few_assignments(lowercase					)					)
   return results




def 						SCREAMING_SNAKE_CASE__ (   lowercase						,lowercase						,lowercase					)  ->       Dict:
   if not check_uniques(lowercase						,lowercase					):
      return False
   elif example["autogenerated"]:
      return False
   elif example["line_max"] > args.line_max:
      return False
   elif example["line_mean"] > args.line_mean:
      return False
   elif example["alpha_frac"] < args.alpha_frac:
      return False
   elif example["ratio"] < args.min_token_ratio:
      return False
   elif example["config_or_test"] and np.random.rand() <= args.filter_proba:
      return False
   elif example["has_no_keywords"] and np.random.rand() <= args.filter_proba:
      return False
   elif example["has_few_assignments"]:
      return False
   else:
      return True




def 						SCREAMING_SNAKE_CASE__ (   lowercase					)  ->       str:
   with open(lowercase						,"""rb"""					) as f_in:
      with gzip.open(str(lowercase					) + """.gz"""						,"""wb"""						,compresslevel=6					) as f_out:
         shutil.copyfileobj(lowercase						,lowercase					)
   os.unlink(lowercase					)


# Settings
lowerCamelCase	:	str				=   HfArgumentParser(PreprocessingArguments)
lowerCamelCase	:	Dict				=   parser.parse_args()
if args.num_workers is None:
       lowerCamelCase	:	Tuple				=   multiprocessing.cpu_count()
lowerCamelCase	:	str				=   AutoTokenizer.from_pretrained(args.tokenizer_dir)

# Load dataset
lowerCamelCase	:	Optional[int]				=   time.time()
lowerCamelCase	:	List[str]				=   load_dataset(args.dataset_name, split='train')
print(f"""Time to load dataset: {time.time()-t_start:.2f}""")

# Run preprocessing
lowerCamelCase	:	List[Any]				=   time.time()
lowerCamelCase	:	Tuple				=   ds.map(preprocess, num_proc=args.num_workers)
print(f"""Time to preprocess dataset: {time.time()-t_start:.2f}""")

# Deduplicate hashes
lowerCamelCase	:	List[str]				=   set(ds.unique('hash'))
lowerCamelCase	:	Tuple				=   len(uniques) / len(ds)
print(f"""Fraction of duplicates: {1-frac:.2%}""")

# Deduplicate data and apply heuristics
lowerCamelCase	:	List[str]				=   time.time()
lowerCamelCase	:	Optional[Any]				=   ds.filter(filter, fn_kwargs={'uniques': uniques, 'args': args})
print(f"""Time to filter dataset: {time.time()-t_start:.2f}""")
print(f"""Size of filtered dataset: {len(ds_filter)}""")

# Deduplicate with minhash and jaccard similarity
if args.near_deduplication:
       lowerCamelCase	:	Tuple				=   time.time()
       lowerCamelCase	,       lowerCamelCase	:	List[Any]				=   deduplicate_dataset(ds_filter, args.jaccard_threshold)
       print(f"""Time to deduplicate dataset: {time.time()-t_start:.2f}""")
       print(f"""Size of deduplicate dataset: {len(ds_filter)}""")

# Save data in batches of samples_per_file
lowerCamelCase	:	Optional[int]				=   Path(args.output_dir)
output_dir.mkdir(exist_ok=True)

# save duplicate_clusters in the output_dir as artifacts
# not sure it is the right place the save it
if args.near_deduplication:
       with open(output_dir / 'duplicate_clusters.json', 'w') as f:
              json.dump(duplicate_clusters, f)

lowerCamelCase	:	Union[str, Any]				=   output_dir / 'data'
data_dir.mkdir(exist_ok=True)

lowerCamelCase	:	Optional[int]				=   time.time()
for file_number, index in enumerate(range(0, len(ds_filter), args.samples_per_file)):
       lowerCamelCase	:	List[str]				=   str(data_dir / f"""file-{file_number+1:012}.json""")
       lowerCamelCase	:	Dict				=   min(len(ds_filter), index + args.samples_per_file)
       ds_filter.select(list(range(index, end_index))).to_json(file_path)
       compress_file(file_path)
print(f"""Time to save dataset: {time.time()-t_start:.2f}""")











import inspect
from typing import Optional, Union

import numpy as np
import PIL
import torch
from torch.nn import functional as F
from torchvision import transforms
from transformers import CLIPFeatureExtractor, CLIPModel, CLIPTextModel, CLIPTokenizer

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    DiffusionPipeline,
    DPMSolverMultistepScheduler,
    LMSDiscreteScheduler,
    PNDMScheduler,
    UNetaDConditionModel,
)
from diffusers.pipelines.stable_diffusion.pipeline_stable_diffusion import StableDiffusionPipelineOutput
from diffusers.utils import (
    PIL_INTERPOLATION,
    randn_tensor,
)




def 						SCREAMING_SNAKE_CASE__ (   lowercase						,lowercase						,lowercase					)  ->       Union[str, Any]:
   if isinstance(lowercase						,torch.Tensor					):
      return image
   elif isinstance(lowercase						,PIL.Image.Image					):
      snake_case				:    str           =	[image]

   if isinstance(image[0]						,PIL.Image.Image					):
      snake_case				:    List[Any]           =	[np.array(i.resize((w, h)						,resample=PIL_INTERPOLATION["""lanczos"""]					)					)[None, :] for i in image]
      snake_case				:    Optional[int]           =	np.concatenate(lowercase						,axis=0					)
      snake_case				:    str           =	np.array(lowercase					).astype(np.floataa					) / 255.0
      snake_case				:    List[str]           =	image.transpose(0						,3						,1						,2					)
      snake_case				:    Any           =	2.0 * image - 1.0
      snake_case				:    Optional[Any]           =	torch.from_numpy(lowercase					)
   elif isinstance(image[0]						,torch.Tensor					):
      snake_case				:    Optional[int]           =	torch.cat(lowercase						,dim=0					)
   return image




def 						SCREAMING_SNAKE_CASE__ (   lowercase						,lowercase						,lowercase						,lowercase=0.9995					)  ->       Optional[int]:
   if not isinstance(lowercase						,np.ndarray					):
      snake_case				:    Any           =	True
      snake_case				:    str           =	va.device
      snake_case				:    Optional[Any]           =	va.cpu().numpy()
      snake_case				:    str           =	va.cpu().numpy()

   snake_case				:    Tuple           =	np.sum(va * va / (np.linalg.norm(lowercase					) * np.linalg.norm(lowercase					))					)
   if np.abs(lowercase					) > DOT_THRESHOLD:
      snake_case				:    Optional[int]           =	(1 - t) * va + t * va
   else:
      snake_case				:    List[Any]           =	np.arccos(lowercase					)
      snake_case				:    str           =	np.sin(lowercase					)
      snake_case				:    int           =	theta_a * t
      snake_case				:    Dict           =	np.sin(lowercase					)
      snake_case				:    Optional[Any]           =	np.sin(theta_a - theta_t					) / sin_theta_a
      snake_case				:    Union[str, Any]           =	sin_theta_t / sin_theta_a
      snake_case				:    Union[str, Any]           =	sa * va + sa * va

   if inputs_are_torch:
      snake_case				:    List[Any]           =	torch.from_numpy(lowercase					).to(lowercase					)

   return va




def 						SCREAMING_SNAKE_CASE__ (   lowercase						,lowercase					)  ->       Tuple:
   snake_case				:    Dict           =	F.normalize(lowercase						,dim=-1					)
   snake_case				:    Optional[Any]           =	F.normalize(lowercase						,dim=-1					)
   return (x - y).norm(dim=-1					).div(2					).arcsin().pow(2					).mul(2					)




def 						SCREAMING_SNAKE_CASE__ (   lowercase						,lowercase					)  ->       str:
   for param in model.parameters():
      snake_case				:    Tuple           =	value




class 						__lowercase		(UpperCamelCase__							):




      """simple docstring"""





      def __init__(				self    ,				A    ,				A    ,				A    ,				A    ,				A    ,				A    ,				A    ,				A=None    ,				A=None    ,				A=None    ,				)		->    List[Any]:
         super().__init__()
         self.register_modules(
             vae=A    ,				text_encoder=A    ,				clip_model=A    ,				tokenizer=A    ,				unet=A    ,				scheduler=A    ,				feature_extractor=A    ,				coca_model=A    ,				coca_tokenizer=A    ,				coca_transform=A    ,				)
         snake_case				:    Optional[int]           =	(
             feature_extractor.size
             if isinstance(feature_extractor.size    ,				A     )
             else feature_extractor.size["""shortest_edge"""]
         )
         snake_case				:    Union[str, Any]           =	transforms.Normalize(mean=feature_extractor.image_mean    ,				std=feature_extractor.image_std     )
         set_requires_grad(self.text_encoder    ,				A     )
         set_requires_grad(self.clip_model    ,				A     )

      def      UpperCAmelCase				(				self    ,				A = "auto"     )		->    Tuple:
         if slice_size == "auto":
            # half the attention head size is usually a good trade-off between
            # speed and memory
            snake_case				:    List[str]           =	self.unet.config.attention_head_dim // 2
         self.unet.set_attention_slice(A     )

      def      UpperCAmelCase				(				self     )		->    Optional[int]:
         self.enable_attention_slicing(A     )

      def      UpperCAmelCase				(				self     )		->    Any:
         set_requires_grad(self.vae    ,				A     )

      def      UpperCAmelCase				(				self     )		->    List[Any]:
         set_requires_grad(self.vae    ,				A     )

      def      UpperCAmelCase				(				self     )		->    Union[str, Any]:
         set_requires_grad(self.unet    ,				A     )

      def      UpperCAmelCase				(				self     )		->    Tuple:
         set_requires_grad(self.unet    ,				A     )

      def      UpperCAmelCase				(				self    ,				A    ,				A    ,				A     )		->    Dict:
         # get the original timestep using init_timestep
         snake_case				:    Tuple           =	min(int(num_inference_steps * strength     )    ,				A     )

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

         return timesteps, num_inference_steps - t_start

      def      UpperCAmelCase				(				self    ,				A    ,				A    ,				A    ,				A    ,				A    ,				A=None     )		->    List[str]:
         if not isinstance(A    ,				torch.Tensor     ):
            raise ValueError(f"""`image` has to be of type `torch.Tensor` but is {type(A     )}"""     )

         snake_case				:    str           =	image.to(device=A    ,				dtype=A     )

         if isinstance(A    ,				A     ):
            snake_case				:    int           =	[
                self.vae.encode(image[i : i + 1]     ).latent_dist.sample(generator[i]     ) for i in range(A     )
            ]
            snake_case				:    str           =	torch.cat(A    ,				dim=0     )
         else:
            snake_case				:    List[Any]           =	self.vae.encode(A     ).latent_dist.sample(A     )

         # Hardcode 0.18215 because stable-diffusion-2-base has not self.vae.config.scaling_factor
         snake_case				:    Dict           =	0.1_82_15 * init_latents
         snake_case				:    Tuple           =	init_latents.repeat_interleave(A    ,				dim=0     )

         snake_case				:    Optional[int]           =	randn_tensor(init_latents.shape    ,				generator=A    ,				device=A    ,				dtype=A     )

         # get latents
         snake_case				:    Union[str, Any]           =	self.scheduler.add_noise(A    ,				A    ,				A     )
         snake_case				:    List[Any]           =	init_latents

         return latents

      def      UpperCAmelCase				(				self    ,				A     )		->    int:
         snake_case				:    Optional[Any]           =	self.coca_transform(A     ).unsqueeze(0     )
         with torch.no_grad(), torch.cuda.amp.autocast():
            snake_case				:    Optional[Any]           =	self.coca_model.generate(transformed_image.to(device=self.device    ,				dtype=self.coca_model.dtype     )     )
         snake_case				:    int           =	self.coca_tokenizer.decode(generated[0].cpu().numpy()     )
         return generated.split("""<end_of_text>"""     )[0].replace("""<start_of_text>"""    ,				""""""     ).rstrip(""" .,"""     )

      def      UpperCAmelCase				(				self    ,				A    ,				A     )		->    List[Any]:
         snake_case				:    Tuple           =	self.feature_extractor.preprocess(A     )
         snake_case				:    List[Any]           =	torch.from_numpy(clip_image_input["""pixel_values"""][0]     ).unsqueeze(0     ).to(self.device     ).half()
         snake_case				:    Optional[int]           =	self.clip_model.get_image_features(A     )
         snake_case				:    List[str]           =	image_embeddings_clip / image_embeddings_clip.norm(p=2    ,				dim=-1    ,				keepdim=A     )
         snake_case				:    Tuple           =	image_embeddings_clip.repeat_interleave(A    ,				dim=0     )
         return image_embeddings_clip

      @torch.enable_grad()
      def      UpperCAmelCase				(				self    ,				A    ,				A    ,				A    ,				A    ,				A    ,				A    ,				A    ,				)		->    Any:
         snake_case				:    Dict           =	latents.detach().requires_grad_()

         snake_case				:    str           =	self.scheduler.scale_model_input(A    ,				A     )

         # predict the noise residual
         snake_case				:    str           =	self.unet(A    ,				A    ,				encoder_hidden_states=A     ).sample

         if isinstance(self.scheduler    ,				(PNDMScheduler, DDIMScheduler, DPMSolverMultistepScheduler)     ):
            snake_case				:    int           =	self.scheduler.alphas_cumprod[timestep]
            snake_case				:    Tuple           =	1 - alpha_prod_t
            # compute predicted original sample from predicted noise also called
            # "predicted x_0" of formula (12) from https://arxiv.org/pdf/2010.02502.pdf
            snake_case				:    Any           =	(latents - beta_prod_t ** 0.5 * noise_pred) / alpha_prod_t ** 0.5

            snake_case				:    str           =	torch.sqrt(A     )
            snake_case				:    str           =	pred_original_sample * (fac) + latents * (1 - fac)
         elif isinstance(self.scheduler    ,				A     ):
            snake_case				:    int           =	self.scheduler.sigmas[index]
            snake_case				:    List[Any]           =	latents - sigma * noise_pred
         else:
            raise ValueError(f"""scheduler type {type(self.scheduler     )} not supported"""     )

         # Hardcode 0.18215 because stable-diffusion-2-base has not self.vae.config.scaling_factor
         snake_case				:    List[str]           =	1 / 0.1_82_15 * sample
         snake_case				:    str           =	self.vae.decode(A     ).sample
         snake_case				:    Any           =	(image / 2 + 0.5).clamp(0    ,				1     )

         snake_case				:    str           =	transforms.Resize(self.feature_extractor_size     )(A     )
         snake_case				:    Dict           =	self.normalize(A     ).to(latents.dtype     )

         snake_case				:    Union[str, Any]           =	self.clip_model.get_image_features(A     )
         snake_case				:    List[str]           =	image_embeddings_clip / image_embeddings_clip.norm(p=2    ,				dim=-1    ,				keepdim=A     )

         snake_case				:    Optional[int]           =	spherical_dist_loss(A    ,				A     ).mean() * clip_guidance_scale

         snake_case				:    int           =	-torch.autograd.grad(A    ,				A     )[0]

         if isinstance(self.scheduler    ,				A     ):
            snake_case				:    Union[str, Any]           =	latents.detach() + grads * (sigma**2)
            snake_case				:    Union[str, Any]           =	noise_pred_original
         else:
            snake_case				:    List[str]           =	noise_pred_original - torch.sqrt(A     ) * grads
         return noise_pred, latents

      @torch.no_grad()
      def __call__(				self    ,				A    ,				A    ,				A = None    ,				A = None    ,				A = 5_1_2    ,				A = 5_1_2    ,				A = 0.6    ,				A = 5_0    ,				A = 7.5    ,				A = 1    ,				A = 0.0    ,				A = 1_0_0    ,				A = None    ,				A = "pil"    ,				A = True    ,				A = 0.8    ,				A = 0.1    ,				A = 0.1    ,				)		->    Union[str, Any]:
         if isinstance(A    ,				A     ) and len(A     ) != batch_size:
            raise ValueError(f"""You have passed {batch_size} batch_size, but only {len(A     )} generators."""     )

         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 isinstance(A    ,				torch.Generator     ) and batch_size > 1:
            snake_case				:    Dict           =	[generator] + [None] * (batch_size - 1)

         snake_case				:    Tuple           =	[
             ("""model""", self.coca_model is None),
             ("""tokenizer""", self.coca_tokenizer is None),
             ("""transform""", self.coca_transform is None),
         ]
         snake_case				:    List[str]           =	[x[0] for x in coca_is_none if x[1]]
         snake_case				:    Optional[int]           =	""", """.join(A     )
         # generate prompts with coca model if prompt is None
         if content_prompt is None:
            if len(A     ):
               raise ValueError(
                   f"""Content prompt is None and CoCa [{coca_is_none_str}] is None."""
                   f"""Set prompt or pass Coca [{coca_is_none_str}] to DiffusionPipeline."""     )
            snake_case				:    Tuple           =	self.get_image_description(A     )
         if style_prompt is None:
            if len(A     ):
               raise ValueError(
                   f"""Style prompt is None and CoCa [{coca_is_none_str}] is None."""
                   f""" Set prompt or pass Coca [{coca_is_none_str}] to DiffusionPipeline."""     )
            snake_case				:    List[Any]           =	self.get_image_description(A     )

         # get prompt text embeddings for content and style
         snake_case				:    Dict           =	self.tokenizer(
             A    ,				padding="""max_length"""    ,				max_length=self.tokenizer.model_max_length    ,				truncation=A    ,				return_tensors="""pt"""    ,				)
         snake_case				:    str           =	self.text_encoder(content_text_input.input_ids.to(self.device     )     )[0]

         snake_case				:    Dict           =	self.tokenizer(
             A    ,				padding="""max_length"""    ,				max_length=self.tokenizer.model_max_length    ,				truncation=A    ,				return_tensors="""pt"""    ,				)
         snake_case				:    Tuple           =	self.text_encoder(style_text_input.input_ids.to(self.device     )     )[0]

         snake_case				:    List[str]           =	slerp(A    ,				A    ,				A     )

         # duplicate text embeddings for each generation per prompt
         snake_case				:    List[Any]           =	text_embeddings.repeat_interleave(A    ,				dim=0     )

         # set timesteps
         snake_case				:    Union[str, Any]           =	"""offset""" in set(inspect.signature(self.scheduler.set_timesteps     ).parameters.keys()     )
         snake_case				:    Optional[Any]           =	{}
         if accepts_offset:
            snake_case				:    Dict           =	1

         self.scheduler.set_timesteps(A    ,				**A     )
         # Some schedulers like PNDM have timesteps as arrays
         # It's more optimized to move all timesteps to correct device beforehand
         self.scheduler.timesteps.to(self.device     )

         snake_case     , snake_case				:    List[Any]           =	self.get_timesteps(A    ,				A    ,				self.device     )
         snake_case				:    List[str]           =	timesteps[:1].repeat(A     )

         # Preprocess image
         snake_case				:    Dict           =	preprocess(A    ,				A    ,				A     )
         snake_case				:    List[Any]           =	self.prepare_latents(
             A    ,				A    ,				A    ,				text_embeddings.dtype    ,				self.device    ,				A     )

         snake_case				:    Optional[int]           =	preprocess(A    ,				A    ,				A     )
         snake_case				:    Optional[Any]           =	self.prepare_latents(
             A    ,				A    ,				A    ,				text_embeddings.dtype    ,				self.device    ,				A     )

         snake_case				:    str           =	slerp(A    ,				A    ,				A     )

         if clip_guidance_scale > 0:
            snake_case				:    List[Any]           =	self.get_clip_image_embeddings(A    ,				A     )
            snake_case				:    Any           =	self.get_clip_image_embeddings(A    ,				A     )
            snake_case				:    Tuple           =	slerp(
                A    ,				A    ,				A     )

         # 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.
         snake_case				:    Optional[Any]           =	guidance_scale > 1.0
         # get unconditional embeddings for classifier free guidance
         if do_classifier_free_guidance:
            snake_case				:    List[str]           =	content_text_input.input_ids.shape[-1]
            snake_case				:    Any           =	self.tokenizer([""""""]    ,				padding="""max_length"""    ,				max_length=A    ,				return_tensors="""pt"""     )
            snake_case				:    Tuple           =	self.text_encoder(uncond_input.input_ids.to(self.device     )     )[0]
            # duplicate unconditional embeddings for each generation per prompt
            snake_case				:    Tuple           =	uncond_embeddings.repeat_interleave(A    ,				dim=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
            snake_case				:    str           =	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`.
         snake_case				:    Tuple           =	(batch_size, self.unet.config.in_channels, height // 8, width // 8)
         snake_case				:    List[Any]           =	text_embeddings.dtype
         if latents is None:
            if self.device.type == "mps":
               # randn does not work reproducibly on mps
               snake_case				:    List[Any]           =	torch.randn(A    ,				generator=A    ,				device="""cpu"""    ,				dtype=A     ).to(
                   self.device     )
            else:
               snake_case				:    Optional[int]           =	torch.randn(A    ,				generator=A    ,				device=self.device    ,				dtype=A     )
         else:
            if latents.shape != latents_shape:
               raise ValueError(f"""Unexpected latents shape, got {latents.shape}, expected {latents_shape}"""     )
            snake_case				:    Union[str, Any]           =	latents.to(self.device     )

         # scale the initial noise by the standard deviation required by the scheduler
         snake_case				:    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]
         snake_case				:    Tuple           =	"""eta""" in set(inspect.signature(self.scheduler.step     ).parameters.keys()     )
         snake_case				:    Dict           =	{}
         if accepts_eta:
            snake_case				:    Union[str, Any]           =	eta

         # check if the scheduler accepts generator
         snake_case				:    List[Any]           =	"""generator""" in set(inspect.signature(self.scheduler.step     ).parameters.keys()     )
         if accepts_generator:
            snake_case				:    List[str]           =	generator

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

               # predict the noise residual
               snake_case				:    Any           =	self.unet(A    ,				A    ,				encoder_hidden_states=A     ).sample

               # perform classifier free guidance
               if do_classifier_free_guidance:
                  snake_case     , snake_case				:    int           =	noise_pred.chunk(2     )
                  snake_case				:    Tuple           =	noise_pred_uncond + guidance_scale * (noise_pred_text - noise_pred_uncond)

               # perform clip guidance
               if clip_guidance_scale > 0:
                  snake_case				:    Any           =	(
                      text_embeddings.chunk(2     )[1] if do_classifier_free_guidance else text_embeddings
                  )
                  snake_case     , snake_case				:    List[Any]           =	self.cond_fn(
                      A    ,				A    ,				A    ,				A    ,				A    ,				A    ,				A    ,				)

               # compute the previous noisy sample x_t -> x_t-1
               snake_case				:    Tuple           =	self.scheduler.step(A    ,				A    ,				A    ,				**A     ).prev_sample

        # Hardcode 0.18215 because stable-diffusion-2-base has not self.vae.config.scaling_factor
         snake_case				:    str           =	1 / 0.1_82_15 * latents
         snake_case				:    Optional[Any]           =	self.vae.decode(A     ).sample

         snake_case				:    Union[str, Any]           =	(image / 2 + 0.5).clamp(0    ,				1     )
         snake_case				:    str           =	image.cpu().permute(0    ,				2    ,				3    ,				1     ).numpy()

         if output_type == "pil":
            snake_case				:    Tuple           =	self.numpy_to_pil(A     )

         if not return_dict:
            return (image, None)

         return StableDiffusionPipelineOutput(images=A    ,				nsfw_content_detected=A     )





