




import os
from glob import glob

import imageio
import torch
import torchvision
import wandb
from img_processing import custom_to_pil, loop_post_process, preprocess, preprocess_vqgan
from loaders import load_vqgan
from PIL import Image
from torch import nn

from transformers import CLIPModel, CLIPTokenizerFast
from utils import get_device, get_timestamp, show_pil
class 						UpperCAmelCase  :







						'''simple docstring'''





						def __init__(		self				:	Dict			,      __lowercase				:	str = "cpu"			,      __lowercase				:	str = "openai/clip-vit-large-patch14"		):






							"""simple docstring"""






							snake_case_						       =	device
							snake_case_						       =	CLIPTokenizerFast.from_pretrained(__lowercase		)
							snake_case_						       =	[0.4814_5466, 0.457_8275, 0.4082_1073]
							snake_case_						       =	[0.2686_2954, 0.2613_0258, 0.2757_7711]
							snake_case_						       =	torchvision.transforms.Normalize(self.image_mean			,      self.image_std		)
							snake_case_						       =	torchvision.transforms.Resize(2_24		)
							snake_case_						       =	torchvision.transforms.CenterCrop(2_24		)






						def 			snake_case__		(		self				:	Any			,      __lowercase				:	Optional[Any]		):






							"""simple docstring"""






							snake_case_						       =	self.resize(__lowercase		)
							snake_case_						       =	self.center_crop(__lowercase		)
							snake_case_						       =	self.normalize(__lowercase		)
							return images






						def __call__(		self				:	List[Any]			,      __lowercase				:	str=None			,      __lowercase				:	Dict=None			,      **__lowercase				:	str		):






							"""simple docstring"""






							snake_case_						       =	self.tokenizer(text=__lowercase			,      **__lowercase		)
							snake_case_						       =	self.preprocess_img(__lowercase		)
							snake_case_						       =	{key: value.to(self.device		) for (key, value) in encoding.items()}
							return encoding







class 						UpperCAmelCase  (							nn.Module     ):







						'''simple docstring'''





						def __init__(		self				:	Union[str, Any]			,      __lowercase				:	Tuple=10			,      __lowercase				:	List[Any]=0.01			,      __lowercase				:	List[str]=None			,      __lowercase				:	Optional[int]=None			,      __lowercase				:	List[str]=None			,      __lowercase				:	List[str]=None			,      __lowercase				:	Union[str, Any]=None			,      __lowercase				:	Dict=None			,      __lowercase				:	str=False			,      __lowercase				:	str=True			,      __lowercase				:	int="image"			,      __lowercase				:	Optional[int]=True			,      __lowercase				:	Optional[int]=False			,      __lowercase				:	int=False			,      __lowercase				:	List[Any]=False			,      ):






							"""simple docstring"""






							super().__init__()
							snake_case_						       =	None
							snake_case_						       =	device if device else get_device()
							if vqgan:
								snake_case_						       =	vqgan
							else:
								snake_case_						       =	load_vqgan(self.device			,      conf_path=__lowercase			,      ckpt_path=__lowercase		)
							self.vqgan.eval()
							if clip:
								snake_case_						       =	clip
							else:
								snake_case_						       =	CLIPModel.from_pretrained("openai/clip-vit-base-patch32"		)
							self.clip.to(self.device		)
							snake_case_						       =	ProcessorGradientFlow(device=self.device		)

							snake_case_						       =	iterations
							snake_case_						       =	lr
							snake_case_						       =	log
							snake_case_						       =	make_grid
							snake_case_						       =	return_val
							snake_case_						       =	quantize
							snake_case_						       =	self.vqgan.decoder.z_shape






						def 			snake_case__		(		self				:	Dict			,      __lowercase				:	Tuple=None			,      __lowercase				:	str=None			,      __lowercase				:	int=5			,      __lowercase				:	Optional[Any]=True		):






							"""simple docstring"""






							snake_case_						       =	[]
							if output_path is None:
								snake_case_						       =	"./animation.gif"
							if input_path is None:
								snake_case_						       =	self.save_path
							snake_case_						       =	sorted(glob(input_path + "/*"		)		)
							if not len(__lowercase		):
								raise ValueError(
								    "No images found in save path, aborting (did you pass save_intermediate=True to the generate"
								    " function?)"		)
							if len(__lowercase		) == 1:
								print("Only one image found in save path, (did you pass save_intermediate=True to the generate function?)"		)
							snake_case_						       =	total_duration / len(__lowercase		)
							snake_case_						       =	[frame_duration] * len(__lowercase		)
							if extend_frames:
								snake_case_						       =	1.5
								snake_case_						       =	3
							for file_name in paths:
								if file_name.endswith(".png"		):
									images.append(imageio.imread(__lowercase		)		)
							imageio.mimsave(__lowercase			,      __lowercase			,      duration=__lowercase		)
							print(f"gif saved to {output_path}"		)






						def 			snake_case__		(		self				:	Dict			,      __lowercase				:	Any=None			,      __lowercase				:	Any=None		):






							"""simple docstring"""






							if not (path or img):
								raise ValueError("Input either path or tensor"		)
							if img is not None:
								raise NotImplementedError
							snake_case_						       =	preprocess(Image.open(__lowercase		)			,      target_image_size=2_56		).to(self.device		)
							snake_case_						       =	preprocess_vqgan(__lowercase		)
							snake_case_       , *snake_case_						       =	self.vqgan.encode(__lowercase		)
							return z






						def 			snake_case__		(		self				:	Tuple			,      __lowercase				:	Union[str, Any]		):






							"""simple docstring"""






							snake_case_						       =	self.latent.detach().requires_grad_()
							snake_case_						       =	base_latent + transform_vector
							if self.quantize:
								snake_case_       , *snake_case_						       =	self.vqgan.quantize(__lowercase		)
							else:
								snake_case_						       =	trans_latent
							return self.vqgan.decode(__lowercase		)






						def 			snake_case__		(		self				:	int			,      __lowercase				:	List[Any]			,      __lowercase				:	Optional[Any]			,      __lowercase				:	List[Any]=None		):






							"""simple docstring"""






							snake_case_						       =	self.clip_preprocessor(text=__lowercase			,      images=__lowercase			,      return_tensors="pt"			,      padding=__lowercase		)
							snake_case_						       =	self.clip(**__lowercase		)
							snake_case_						       =	clip_outputs.logits_per_image
							if weights is not None:
								snake_case_						       =	similarity_logits * weights
							return similarity_logits.sum()






						def 			snake_case__		(		self				:	Any			,      __lowercase				:	Union[str, Any]			,      __lowercase				:	Tuple			,      __lowercase				:	Union[str, Any]		):






							"""simple docstring"""






							snake_case_						       =	self._get_clip_similarity(pos_prompts["prompts"]			,      __lowercase			,      weights=(1 / pos_prompts["weights"])		)
							if neg_prompts:
								snake_case_						       =	self._get_clip_similarity(neg_prompts["prompts"]			,      __lowercase			,      weights=neg_prompts["weights"]		)
							else:
								snake_case_						       =	torch.tensor([1]			,      device=self.device		)
							snake_case_						       =	-torch.log(__lowercase		) + torch.log(__lowercase		)
							return loss






						def 			snake_case__		(		self				:	Dict			,      __lowercase				:	str			,      __lowercase				:	Optional[int]			,      __lowercase				:	Any		):






							"""simple docstring"""






							snake_case_						       =	torch.randn_like(self.latent			,      requires_grad=__lowercase			,      device=self.device		)
							snake_case_						       =	torch.optim.Adam([vector]			,      lr=self.lr		)

							for i in range(self.iterations		):
								optim.zero_grad()
								snake_case_						       =	self._add_vector(__lowercase		)
								snake_case_						       =	loop_post_process(__lowercase		)
								snake_case_						       =	self._get_CLIP_loss(__lowercase			,      __lowercase			,      __lowercase		)
								print("CLIP loss"			,      __lowercase		)
								if self.log:
									wandb.log({"CLIP Loss": clip_loss}		)
								clip_loss.backward(retain_graph=__lowercase		)
								optim.step()
								if self.return_val == "image":
									yield custom_to_pil(transformed_img[0]		)
								else:
									yield vector






						def 			snake_case__		(		self				:	List[Any]			,      __lowercase				:	List[str]			,      __lowercase				:	List[Any]			,      __lowercase				:	Optional[Any]		):






							"""simple docstring"""






							wandb.init(reinit=__lowercase			,      project="face-editor"		)
							wandb.config.update({"Positive Prompts": positive_prompts}		)
							wandb.config.update({"Negative Prompts": negative_prompts}		)
							wandb.config.update({"lr": self.lr, "iterations": self.iterations}		)
							if image_path:
								snake_case_						       =	Image.open(__lowercase		)
								snake_case_						       =	image.resize((2_56, 2_56)		)
								wandb.log("Original Image"			,      wandb.Image(__lowercase		)		)






						def 			snake_case__		(		self				:	List[str]			,      __lowercase				:	List[Any]		):






							"""simple docstring"""






							if not prompts:
								return []
							snake_case_						       =	[]
							snake_case_						       =	[]
							if isinstance(__lowercase			,      __lowercase		):
								snake_case_						       =	[prompt.strip() for prompt in prompts.split("|"		)]
							for prompt in prompts:
								if isinstance(__lowercase			,      (tuple, list)		):
									snake_case_						       =	prompt[0]
									snake_case_						       =	float(prompt[1]		)
								elif ":" in prompt:
									snake_case_       , snake_case_						       =	prompt.split(":"		)
									snake_case_						       =	float(__lowercase		)
								else:
									snake_case_						       =	prompt
									snake_case_						       =	1.0
								processed_prompts.append(__lowercase		)
								weights.append(__lowercase		)
							return {
							    "prompts": processed_prompts,
							    "weights": torch.tensor(__lowercase			,      device=self.device		),
							}






						def 			snake_case__		(		self				:	Dict			,      __lowercase				:	int			,      __lowercase				:	Optional[int]=None			,      __lowercase				:	int=None			,      __lowercase				:	Dict=True			,      __lowercase				:	str=False			,      __lowercase				:	List[Any]=True			,      __lowercase				:	List[str]=True			,      __lowercase				:	List[str]=None			,      ):






							"""simple docstring"""






							if image_path:
								snake_case_						       =	self._get_latent(__lowercase		)
							else:
								snake_case_						       =	torch.randn(self.latent_dim			,      device=self.device		)
							if self.log:
								self._init_logging(__lowercase			,      __lowercase			,      __lowercase		)

							assert pos_prompts, "You must provide at least one positive prompt."
							snake_case_						       =	self.process_prompts(__lowercase		)
							snake_case_						       =	self.process_prompts(__lowercase		)
							if save_final and save_path is None:
								snake_case_						       =	os.path.join("./outputs/"			,      "_".join(pos_prompts["prompts"]		)		)
							if not os.path.exists(__lowercase		):
								os.makedirs(__lowercase		)
							else:
								snake_case_						       =	save_path + "_" + get_timestamp()
								os.makedirs(__lowercase		)
							snake_case_						       =	save_path

							snake_case_						       =	self.vqgan.decode(self.latent		)[0]
							if show_intermediate:
								print("Original Image"		)
								show_pil(custom_to_pil(__lowercase		)		)

							snake_case_						       =	loop_post_process(__lowercase		)
							for iter, transformed_img in enumerate(self._optimize_CLIP(__lowercase			,      __lowercase			,      __lowercase		)		):
								if show_intermediate:
									show_pil(__lowercase		)
								if save_intermediate:
									transformed_img.save(os.path.join(self.save_path			,      f"iter_{iter:03d}.png"		)		)
								if self.log:
									wandb.log({"Image": wandb.Image(__lowercase		)}		)
							if show_final:
								show_pil(__lowercase		)
							if save_final:
								transformed_img.save(os.path.join(self.save_path			,      f"iter_{iter:03d}_final.png"		)		)






import warnings
from typing import Any, Dict, List, Optional, Union

import numpy as np

from ...audio_utils import mel_filter_bank, optimal_fft_length, spectrogram, window_function
from ...feature_extraction_sequence_utils import SequenceFeatureExtractor
from ...feature_extraction_utils import BatchFeature
from ...utils import PaddingStrategy, TensorType, logging


lowercase__							:	Tuple											=		logging.get_logger(__name__)
class 						UpperCAmelCase  (							UpperCAmelCase__     ):







						'''simple docstring'''





						lowerCAmelCase_							=  ['''input_values''', '''attention_mask''']






						def __init__(		self				:	List[Any]			,      __lowercase				:	int = 1			,      __lowercase				:	int = 1_60_00			,      __lowercase				:	float = 0.0			,      __lowercase				:	bool = False			,      __lowercase				:	int = 80			,      __lowercase				:	int = 16			,      __lowercase				:	int = 64			,      __lowercase				:	str = "hann_window"			,      __lowercase				:	float = 1.0			,      __lowercase				:	float = 80			,      __lowercase				:	float = 76_00			,      __lowercase				:	float = 1E-10			,      __lowercase				:	int = 2			,      __lowercase				:	bool = True			,      **__lowercase				:	Union[str, Any]			,      ):






							"""simple docstring"""






							super().__init__(feature_size=__lowercase			,      sampling_rate=__lowercase			,      padding_value=__lowercase			,      **__lowercase		)
							snake_case_						       =	do_normalize
							snake_case_						       =	return_attention_mask

							snake_case_						       =	num_mel_bins
							snake_case_						       =	hop_length
							snake_case_						       =	win_length
							snake_case_						       =	win_function
							snake_case_						       =	frame_signal_scale
							snake_case_						       =	fmin
							snake_case_						       =	fmax
							snake_case_						       =	mel_floor
							snake_case_						       =	reduction_factor

							snake_case_						       =	win_length * sampling_rate // 10_00
							snake_case_						       =	hop_length * sampling_rate // 10_00
							snake_case_						       =	optimal_fft_length(self.sample_size		)
							snake_case_						       =	(self.n_fft // 2) + 1

							snake_case_						       =	window_function(window_length=self.sample_size			,      name=self.win_function			,      periodic=__lowercase		)

							snake_case_						       =	mel_filter_bank(
							    num_frequency_bins=self.n_freqs			,      num_mel_filters=self.num_mel_bins			,      min_frequency=self.fmin			,      max_frequency=self.fmax			,      sampling_rate=self.sampling_rate			,      norm="slaney"			,      mel_scale="slaney"			,      )

							if frame_signal_scale != 1.0:
								warnings.warn(
								    "The argument `frame_signal_scale` is deprecated and will be removed in version 4.30.0 of Transformers"			,      __lowercase			,      )
							if reduction_factor != 2.0:
								warnings.warn(
								    "The argument `reduction_factor` is deprecated and will be removed in version 4.30.0 of Transformers"			,      __lowercase			,      )






						@staticmethod
						# Copied from transformers.models.wav2vec2.feature_extraction_wav2vec2.Wav2Vec2FeatureExtractor.zero_mean_unit_var_norm
						def 			snake_case__		(		__lowercase				:	List[np.ndarray]			,      __lowercase				:	List[np.ndarray]			,      __lowercase				:	float = 0.0		):






							"""simple docstring"""






							if attention_mask is not None:
								snake_case_						       =	np.array(__lowercase			,      np.intaa		)
								snake_case_						       =	[]

								for vector, length in zip(__lowercase			,      attention_mask.sum(-1		)		):
									snake_case_						       =	(vector - vector[:length].mean()) / np.sqrt(vector[:length].var() + 1E-7		)
									if length < normed_slice.shape[0]:
										snake_case_						       =	padding_value

									normed_input_values.append(__lowercase		)
							else:
								snake_case_						       =	[(x - x.mean()) / np.sqrt(x.var() + 1E-7		) for x in input_values]

							return normed_input_values






						def 			snake_case__		(		self				:	str			,      __lowercase				:	np.ndarray			,      ):






							"""simple docstring"""






							snake_case_						       =	spectrogram(
							    __lowercase			,      window=self.window			,      frame_length=self.sample_size			,      hop_length=self.sample_stride			,      fft_length=self.n_fft			,      mel_filters=self.mel_filters			,      mel_floor=self.mel_floor			,      log_mel="log10"			,      )
							return log_mel_spec.T






						def __call__(		self				:	Optional[int]			,      __lowercase				:	Optional[Union[np.ndarray, List[float], List[np.ndarray], List[List[float]]]] = None			,      __lowercase				:	Optional[Union[np.ndarray, List[float], List[np.ndarray], List[List[float]]]] = None			,      __lowercase				:	Union[bool, str, PaddingStrategy] = False			,      __lowercase				:	Optional[int] = None			,      __lowercase				:	bool = False			,      __lowercase				:	Optional[int] = None			,      __lowercase				:	Optional[bool] = None			,      __lowercase				:	Optional[Union[str, TensorType]] = None			,      __lowercase				:	Optional[int] = None			,      **__lowercase				:	List[str]			,      ):






							"""simple docstring"""






							if audio is None and audio_target is None:
								raise ValueError("You must provide either `audio` or `audio_target` values."		)

							if sampling_rate is not None:
								if sampling_rate != self.sampling_rate:
									raise ValueError(
									    f"The model corresponding to this feature extractor: {self} was trained using a sampling rate of"
									    f" {self.sampling_rate}. Please make sure that the provided audio input was sampled with"
									    f" {self.sampling_rate} and not {sampling_rate}."		)
							else:
								logger.warning(
								    "It is strongly recommended to pass the ``sampling_rate`` argument to this function. "
								    "Failing to do so can result in silent errors that might be hard to debug."		)

							if audio is not None:
								snake_case_						       =	self._process_audio(
								    __lowercase			,      __lowercase			,      __lowercase			,      __lowercase			,      __lowercase			,      __lowercase			,      __lowercase			,      __lowercase			,      **__lowercase			,      )
							else:
								snake_case_						       =	None

							if audio_target is not None:
								snake_case_						       =	self._process_audio(
								    __lowercase			,      __lowercase			,      __lowercase			,      __lowercase			,      __lowercase			,      __lowercase			,      __lowercase			,      __lowercase			,      **__lowercase			,      )

								if inputs is None:
									return inputs_target
								else:
									snake_case_						       =	inputs_target["input_values"]
									snake_case_						       =	inputs_target.get("attention_mask"		)
									if decoder_attention_mask is not None:
										snake_case_						       =	decoder_attention_mask

							return inputs






						def 			snake_case__		(		self				:	Dict			,      __lowercase				:	Union[np.ndarray, List[float], List[np.ndarray], List[List[float]]]			,      __lowercase				:	bool = False			,      __lowercase				:	Union[bool, str, PaddingStrategy] = False			,      __lowercase				:	Optional[int] = None			,      __lowercase				:	bool = False			,      __lowercase				:	Optional[int] = None			,      __lowercase				:	Optional[bool] = None			,      __lowercase				:	Optional[Union[str, TensorType]] = None			,      **__lowercase				:	Optional[Any]			,      ):






							"""simple docstring"""






							snake_case_						       =	isinstance(__lowercase			,      np.ndarray		) and len(speech.shape		) > 1
							if is_batched_numpy and len(speech.shape		) > 2:
								raise ValueError(f"Only mono-channel audio is supported for input to {self}"		)
							snake_case_						       =	is_batched_numpy or (
							    isinstance(__lowercase			,      (list, tuple)		) and (isinstance(speech[0]			,      (np.ndarray, tuple, list)		))
							)

							if is_batched:
								snake_case_						       =	[np.asarray(__lowercase			,      dtype=np.floataa		) for speech in speech]
							elif not is_batched and not isinstance(__lowercase			,      np.ndarray		):
								snake_case_						       =	np.asarray(__lowercase			,      dtype=np.floataa		)
							elif isinstance(__lowercase			,      np.ndarray		) and speech.dtype is np.dtype(np.floataa		):
								snake_case_						       =	speech.astype(np.floataa		)

							# always return batch
							if not is_batched:
								snake_case_						       =	[speech]

							# needed to make pad() work on spectrogram inputs
							snake_case_						       =	self.feature_size

							# convert into correct format for padding
							if is_target:
								snake_case_						       =	[self._extract_mel_features(__lowercase		) for waveform in speech]
								snake_case_						       =	BatchFeature({"input_values": features}		)
								snake_case_						       =	self.num_mel_bins
							else:
								snake_case_						       =	BatchFeature({"input_values": speech}		)

							snake_case_						       =	self.pad(
							    __lowercase			,      padding=__lowercase			,      max_length=__lowercase			,      truncation=__lowercase			,      pad_to_multiple_of=__lowercase			,      return_attention_mask=__lowercase			,      **__lowercase			,      )

							snake_case_						       =	feature_size_hack

							# convert input values to correct format
							snake_case_						       =	padded_inputs["input_values"]
							if not isinstance(input_values[0]			,      np.ndarray		):
								snake_case_						       =	[np.asarray(__lowercase			,      dtype=np.floataa		) for array in input_values]
							elif (
							    not isinstance(__lowercase			,      np.ndarray		)
							    and isinstance(input_values[0]			,      np.ndarray		)
							    and input_values[0].dtype is np.dtype(np.floataa		)
							):
								snake_case_						       =	[array.astype(np.floataa		) for array in input_values]
							elif isinstance(__lowercase			,      np.ndarray		) and input_values.dtype is np.dtype(np.floataa		):
								snake_case_						       =	input_values.astype(np.floataa		)

							# convert attention_mask to correct format
							snake_case_						       =	padded_inputs.get("attention_mask"		)
							if attention_mask is not None:
								snake_case_						       =	[np.asarray(__lowercase			,      dtype=np.intaa		) for array in attention_mask]

							# zero-mean and unit-variance normalization
							if not is_target and self.do_normalize:
								snake_case_						       =	(
								    attention_mask
								    if self._get_padding_strategies(__lowercase			,      max_length=__lowercase		) is not PaddingStrategy.DO_NOT_PAD
								    else None
								)
								snake_case_						       =	self.zero_mean_unit_var_norm(
								    padded_inputs["input_values"]			,      attention_mask=__lowercase			,      padding_value=self.padding_value		)

							if return_tensors is not None:
								snake_case_						       =	padded_inputs.convert_to_tensors(__lowercase		)

							return padded_inputs






						def 			snake_case__		(		self				:	Tuple		):






							"""simple docstring"""






							snake_case_						       =	super().to_dict()

							# Don't serialize these as they are derived from the other properties.
							snake_case_						       =	["window", "mel_filters", "sample_size", "sample_stride", "n_fft", "n_freqs"]
							for name in names:
								if name in output:
									del output[name]

							return output
