


"""simple docstring"""







import logging
import os
from dataclasses import dataclass, field
from typing import Dict, Optional

import numpy as np
from utils_multiple_choice import MultipleChoiceDataset, Split, processors

import transformers
from transformers import (
    AutoConfig,
    AutoModelForMultipleChoice,
    AutoTokenizer,
    DataCollatorWithPadding,
    EvalPrediction,
    HfArgumentParser,
    Trainer,
    TrainingArguments,
    set_seed,
)
from transformers.trainer_utils import is_main_process


a__    :							Any            =					logging.getLogger(__name__)




def        UpperCAmelCase__	(lowerCAmelCase_  ,						lowerCAmelCase_       ):




					'''simple docstring'''


					return (preds == labels).mean()




@dataclass
class 			UpperCamelCase_     :

	"""simple docstring"""





	snake_case__    :  str    =      field(
	    metadata={"help": "Path to pretrained model or model identifier from huggingface.co/models"})
	snake_case__    :  Optional[str]    =      field(
	    default=UpperCamelCase      ,  metadata={"help": "Pretrained config name or path if not the same as model_name"})
	snake_case__    :  Optional[str]    =      field(
	    default=UpperCamelCase      ,  metadata={"help": "Pretrained tokenizer name or path if not the same as model_name"})
	snake_case__    :  Optional[str]    =      field(
	    default=UpperCamelCase      ,  metadata={"help": "Where do you want to store the pretrained models downloaded from huggingface.co"}      ,  )




@dataclass
class 			UpperCamelCase_     :

	"""simple docstring"""





	snake_case__    :  str    =      field(metadata={"help": "The name of the task to train on: " + ", ".join(processors.keys())})
	snake_case__    :  str    =      field(metadata={"help": "Should contain the data files for the task."})
	snake_case__    :  int    =      field(
	    default=128      ,  metadata={
	        "help": (
	            "The maximum total input sequence length after tokenization. Sequences longer "
	            "than this will be truncated, sequences shorter will be padded."
	        )
	    }      ,  )
	snake_case__    :  bool    =      field(
	    default=UpperCamelCase      ,  metadata={"help": "Overwrite the cached training and evaluation sets"})




def        UpperCAmelCase__	():




					'''simple docstring'''


					__SCREAMING_SNAKE_CASE						       =    HfArgumentParser((ModelArguments, DataTrainingArguments, TrainingArguments)       )
					__SCREAMING_SNAKE_CASE       ,  __SCREAMING_SNAKE_CASE       ,  __SCREAMING_SNAKE_CASE						       =    parser.parse_args_into_dataclasses()

					if (
					    os.path.exists(training_args.output_dir       )
					    and os.listdir(training_args.output_dir       )
					    and training_args.do_train
					    and not training_args.overwrite_output_dir
					):
										raise ValueError(
										    f"""Output directory ({training_args.output_dir}) already exists and is not empty. Use"""
										    " --overwrite_output_dir to overcome."       )

					# Setup logging
					logging.basicConfig(
					    format="%(asctime)s - %(levelname)s - %(name)s - %(message)s"  ,						datefmt="%m/%d/%Y %H:%M:%S"  ,						level=logging.INFO if training_args.local_rank in [-1, 0] else logging.WARN  ,						)
					logger.warning(
					    "Process rank: %s, device: %s, n_gpu: %s, distributed training: %s, 16-bits training: %s"  ,						training_args.local_rank  ,						training_args.device  ,						training_args.n_gpu  ,						bool(training_args.local_rank != -1       )  ,						training_args.fpaa  ,						)
					# Set the verbosity to info of the Transformers logger (on main process only):
					if is_main_process(training_args.local_rank       ):
										transformers.utils.logging.set_verbosity_info()
										transformers.utils.logging.enable_default_handler()
										transformers.utils.logging.enable_explicit_format()
					logger.info("Training/evaluation parameters %s"  ,						lowerCAmelCase_       )

					# Set seed
					set_seed(training_args.seed       )

					try:
										__SCREAMING_SNAKE_CASE						       =    processors[data_args.task_name]()
										__SCREAMING_SNAKE_CASE						       =    processor.get_labels()
										__SCREAMING_SNAKE_CASE						       =    len(lowerCAmelCase_       )
					except KeyError:
										raise ValueError("Task not found: %s" % (data_args.task_name)       )

					# Load pretrained model and tokenizer
					#
					# Distributed training:
					# The .from_pretrained methods guarantee that only one local process can concurrently
					# download model & vocab.

					__SCREAMING_SNAKE_CASE						       =    AutoConfig.from_pretrained(
					    model_args.config_name if model_args.config_name else model_args.model_name_or_path  ,						num_labels=lowerCAmelCase_  ,						finetuning_task=data_args.task_name  ,						cache_dir=model_args.cache_dir  ,						)
					__SCREAMING_SNAKE_CASE						       =    AutoTokenizer.from_pretrained(
					    model_args.tokenizer_name if model_args.tokenizer_name else model_args.model_name_or_path  ,						cache_dir=model_args.cache_dir  ,						)
					__SCREAMING_SNAKE_CASE						       =    AutoModelForMultipleChoice.from_pretrained(
					    model_args.model_name_or_path  ,						from_tf=bool(".ckpt" in model_args.model_name_or_path       )  ,						config=lowerCAmelCase_  ,						cache_dir=model_args.cache_dir  ,						)

					# Get datasets
					__SCREAMING_SNAKE_CASE						       =    (
					    MultipleChoiceDataset(
					        data_dir=data_args.data_dir  ,						tokenizer=lowerCAmelCase_  ,						task=data_args.task_name  ,						max_seq_length=data_args.max_seq_length  ,						overwrite_cache=data_args.overwrite_cache  ,						mode=Split.train  ,						)
					    if training_args.do_train
					    else None
					)
					__SCREAMING_SNAKE_CASE						       =    (
					    MultipleChoiceDataset(
					        data_dir=data_args.data_dir  ,						tokenizer=lowerCAmelCase_  ,						task=data_args.task_name  ,						max_seq_length=data_args.max_seq_length  ,						overwrite_cache=data_args.overwrite_cache  ,						mode=Split.dev  ,						)
					    if training_args.do_eval
					    else None
					)

					def compute_metrics(lowerCAmelCase_       ) -> Dict:
										__SCREAMING_SNAKE_CASE						       =    np.argmax(p.predictions  ,						axis=1       )
										return {"acc": simple_accuracy(lowerCAmelCase_  ,						p.label_ids       )}

					# Data collator
					__SCREAMING_SNAKE_CASE						       =    DataCollatorWithPadding(lowerCAmelCase_  ,						pad_to_multiple_of=8       ) if training_args.fpaa else None

					# Initialize our Trainer
					__SCREAMING_SNAKE_CASE						       =    Trainer(
					    model=lowerCAmelCase_  ,						args=lowerCAmelCase_  ,						train_dataset=lowerCAmelCase_  ,						eval_dataset=lowerCAmelCase_  ,						compute_metrics=lowerCAmelCase_  ,						data_collator=lowerCAmelCase_  ,						)

					# Training
					if training_args.do_train:
										trainer.train(
										    model_path=model_args.model_name_or_path if os.path.isdir(model_args.model_name_or_path       ) else None       )
										trainer.save_model()
										# For convenience, we also re-save the tokenizer to the same directory,
										# so that you can share your model easily on huggingface.co/models =)
										if trainer.is_world_master():
															tokenizer.save_pretrained(training_args.output_dir       )

    # Evaluation
					__SCREAMING_SNAKE_CASE						       =    {}
					if training_args.do_eval:
										logger.info("*** Evaluate ***"       )

										__SCREAMING_SNAKE_CASE						       =    trainer.evaluate()

										__SCREAMING_SNAKE_CASE						       =    os.path.join(training_args.output_dir  ,						"eval_results.txt"       )
										if trainer.is_world_master():
															with open(lowerCAmelCase_  ,						"w"       ) as writer:
																				logger.info("***** Eval results *****"       )
																				for key, value in result.items():
																									logger.info("  %s = %s"  ,						lowerCAmelCase_  ,						lowerCAmelCase_       )
																									writer.write("%s = %s\n" % (key, value)       )

																				results.update(lowerCAmelCase_       )

					return results




def        UpperCAmelCase__	(lowerCAmelCase_       ):




					'''simple docstring'''


					main()


if __name__ == "__main__":
		main()




"""simple docstring"""







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

import torch
from transformers import (
    CLIPImageProcessor,
    CLIPTextModel,
    CLIPTokenizer,
    WhisperForConditionalGeneration,
    WhisperProcessor,
)

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    DiffusionPipeline,
    LMSDiscreteScheduler,
    PNDMScheduler,
    UNetaDConditionModel,
)
from diffusers.pipelines.stable_diffusion.pipeline_stable_diffusion import StableDiffusionPipelineOutput
from diffusers.pipelines.stable_diffusion.safety_checker import StableDiffusionSafetyChecker
from diffusers.utils import logging


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




class 			UpperCamelCase_     (			UpperCamelCase):

	"""simple docstring"""





	def __init__(			self      :    List[str]      ,      UpperCAmelCase__      :    WhisperForConditionalGeneration      ,      UpperCAmelCase__      :    WhisperProcessor      ,      UpperCAmelCase__      :    AutoencoderKL      ,      UpperCAmelCase__      :    CLIPTextModel      ,      UpperCAmelCase__      :    CLIPTokenizer      ,      UpperCAmelCase__      :    UNetaDConditionModel      ,      UpperCAmelCase__      :    Union[DDIMScheduler, PNDMScheduler, LMSDiscreteScheduler]      ,      UpperCAmelCase__      :    StableDiffusionSafetyChecker      ,      UpperCAmelCase__      :    CLIPImageProcessor      ,      )	->     Optional[int]:
						super().__init__()

						if safety_checker is None:
											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 ."       )

						self.register_modules(
						    speech_model=UpperCAmelCase__      ,      speech_processor=UpperCAmelCase__      ,      vae=UpperCAmelCase__      ,      text_encoder=UpperCAmelCase__      ,      tokenizer=UpperCAmelCase__      ,      unet=UpperCAmelCase__      ,      scheduler=UpperCAmelCase__      ,      feature_extractor=UpperCAmelCase__      ,      )




	def   UpperCAmelCase_				(			self      :    Optional[Any]      ,      UpperCAmelCase__      :    Optional[Union[str, int]] = "auto"       )	->     str:
						if slice_size == "auto":
											__SCREAMING_SNAKE_CASE						       =    self.unet.config.attention_head_dim // 2
						self.unet.set_attention_slice(UpperCAmelCase__       )




	def   UpperCAmelCase_				(			self      :    Tuple       )	->     List[Any]:
						self.enable_attention_slicing(UpperCAmelCase__       )




	@torch.no_grad()
	def __call__(			self      :    str      ,      UpperCAmelCase__      :    Optional[int]      ,      UpperCAmelCase__      :    str=1_6_0_0_0      ,      UpperCAmelCase__      :    int = 5_1_2      ,      UpperCAmelCase__      :    int = 5_1_2      ,      UpperCAmelCase__      :    int = 5_0      ,      UpperCAmelCase__      :    float = 7.5      ,      UpperCAmelCase__      :    Optional[Union[str, List[str]]] = None      ,      UpperCAmelCase__      :    Optional[int] = 1      ,      UpperCAmelCase__      :    float = 0.0      ,      UpperCAmelCase__      :    Optional[torch.Generator] = None      ,      UpperCAmelCase__      :    Optional[torch.FloatTensor] = None      ,      UpperCAmelCase__      :    Optional[str] = "pil"      ,      UpperCAmelCase__      :    bool = True      ,      UpperCAmelCase__      :    Optional[Callable[[int, int, torch.FloatTensor], None]] = None      ,      UpperCAmelCase__      :    int = 1      ,      **UpperCAmelCase__      :    Dict      ,      )	->     Any:
						__SCREAMING_SNAKE_CASE						       =    self.speech_processor.feature_extractor(
						    UpperCAmelCase__      ,      return_tensors="pt"      ,      sampling_rate=UpperCAmelCase__       ).input_features.to(self.device       )
						__SCREAMING_SNAKE_CASE						       =    self.speech_model.generate(UpperCAmelCase__      ,      max_length=4_8_0_0_0_0       )

						__SCREAMING_SNAKE_CASE						       =    self.speech_processor.tokenizer.batch_decode(UpperCAmelCase__      ,      skip_special_tokens=UpperCAmelCase__      ,      normalize=UpperCAmelCase__       )[
						    0
						]

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

						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(UpperCAmelCase__      ,      UpperCAmelCase__       ) or callback_steps <= 0)
						):
											raise ValueError(
											    F"""`callback_steps` has to be a positive integer but is {callback_steps} of type"""
											    F""" {type(UpperCAmelCase__       )}."""       )

						# get prompt text embeddings
						__SCREAMING_SNAKE_CASE						       =    self.tokenizer(
						    UpperCAmelCase__      ,      padding="max_length"      ,      max_length=self.tokenizer.model_max_length      ,      return_tensors="pt"      ,      )
						__SCREAMING_SNAKE_CASE						       =    text_inputs.input_ids

						if text_input_ids.shape[-1] > self.tokenizer.model_max_length:
											__SCREAMING_SNAKE_CASE						       =    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}"""       )
											__SCREAMING_SNAKE_CASE						       =    text_input_ids[:, : self.tokenizer.model_max_length]
						__SCREAMING_SNAKE_CASE						       =    self.text_encoder(text_input_ids.to(self.device       )       )[0]

						# duplicate text embeddings for each generation per prompt, using mps friendly method
						__SCREAMING_SNAKE_CASE       ,  __SCREAMING_SNAKE_CASE       ,  __SCREAMING_SNAKE_CASE						       =    text_embeddings.shape
						__SCREAMING_SNAKE_CASE						       =    text_embeddings.repeat(1      ,      UpperCAmelCase__      ,      1       )
						__SCREAMING_SNAKE_CASE						       =    text_embeddings.view(bs_embed * num_images_per_prompt      ,      UpperCAmelCase__      ,      -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.
						__SCREAMING_SNAKE_CASE						       =    guidance_scale > 1.0
						# get unconditional embeddings for classifier free guidance
						if do_classifier_free_guidance:
											__SCREAMING_SNAKE_CASE						       =    42
											if negative_prompt is None:
																__SCREAMING_SNAKE_CASE						       =    [""] * batch_size
											elif type(UpperCAmelCase__       ) is not type(UpperCAmelCase__       ):
																raise TypeError(
																    F"""`negative_prompt` should be the same type to `prompt`, but got {type(UpperCAmelCase__       )} !="""
																    F""" {type(UpperCAmelCase__       )}."""       )
											elif isinstance(UpperCAmelCase__      ,      UpperCAmelCase__       ):
																__SCREAMING_SNAKE_CASE						       =    [negative_prompt]
											elif batch_size != len(UpperCAmelCase__       ):
																raise ValueError(
																    F"""`negative_prompt`: {negative_prompt} has batch size {len(UpperCAmelCase__       )}, but `prompt`:"""
																    F""" {prompt} has batch size {batch_size}. Please make sure that passed `negative_prompt` matches"""
																    " the batch size of `prompt`."       )
											else:
																__SCREAMING_SNAKE_CASE						       =    negative_prompt

											__SCREAMING_SNAKE_CASE						       =    text_input_ids.shape[-1]
											__SCREAMING_SNAKE_CASE						       =    self.tokenizer(
											    UpperCAmelCase__      ,      padding="max_length"      ,      max_length=UpperCAmelCase__      ,      truncation=UpperCAmelCase__      ,      return_tensors="pt"      ,      )
											__SCREAMING_SNAKE_CASE						       =    self.text_encoder(uncond_input.input_ids.to(self.device       )       )[0]

											# duplicate unconditional embeddings for each generation per prompt, using mps friendly method
											__SCREAMING_SNAKE_CASE						       =    uncond_embeddings.shape[1]
											__SCREAMING_SNAKE_CASE						       =    uncond_embeddings.repeat(1      ,      UpperCAmelCase__      ,      1       )
											__SCREAMING_SNAKE_CASE						       =    uncond_embeddings.view(batch_size * num_images_per_prompt      ,      UpperCAmelCase__      ,      -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
											__SCREAMING_SNAKE_CASE						       =    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`.
						__SCREAMING_SNAKE_CASE						       =    (batch_size * num_images_per_prompt, self.unet.config.in_channels, height // 8, width // 8)
						__SCREAMING_SNAKE_CASE						       =    text_embeddings.dtype
						if latents is None:
											if self.device.type == "mps":
																# randn does not exist on mps
																__SCREAMING_SNAKE_CASE						       =    torch.randn(UpperCAmelCase__      ,      generator=UpperCAmelCase__      ,      device="cpu"      ,      dtype=UpperCAmelCase__       ).to(
																    self.device       )
											else:
																__SCREAMING_SNAKE_CASE						       =    torch.randn(UpperCAmelCase__      ,      generator=UpperCAmelCase__      ,      device=self.device      ,      dtype=UpperCAmelCase__       )
						else:
											if latents.shape != latents_shape:
																raise ValueError(F"""Unexpected latents shape, got {latents.shape}, expected {latents_shape}"""       )
											__SCREAMING_SNAKE_CASE						       =    latents.to(self.device       )

						# set timesteps
						self.scheduler.set_timesteps(UpperCAmelCase__       )

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

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

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

											# predict the noise residual
											__SCREAMING_SNAKE_CASE						       =    self.unet(UpperCAmelCase__      ,      UpperCAmelCase__      ,      encoder_hidden_states=UpperCAmelCase__       ).sample

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

											# compute the previous noisy sample x_t -> x_t-1
											__SCREAMING_SNAKE_CASE						       =    self.scheduler.step(UpperCAmelCase__      ,      UpperCAmelCase__      ,      UpperCAmelCase__      ,      **UpperCAmelCase__       ).prev_sample

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

						__SCREAMING_SNAKE_CASE						       =    1 / 0.18_215 * latents
						__SCREAMING_SNAKE_CASE						       =    self.vae.decode(UpperCAmelCase__       ).sample

						__SCREAMING_SNAKE_CASE						       =    (image / 2 + 0.5).clamp(0      ,      1       )

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

						if output_type == "pil":
											__SCREAMING_SNAKE_CASE						       =    self.numpy_to_pil(UpperCAmelCase__       )

						if not return_dict:
											return image

						return StableDiffusionPipelineOutput(images=UpperCAmelCase__      ,      nsfw_content_detected=UpperCAmelCase__       )
