

from copy import deepcopy
from pathlib import Path
from typing import TYPE_CHECKING, Any, Callable, Dict, List, Optional, Tuple, Union

import torch
from torch import nn
from torch.utils.data import Dataset

from .deepspeed import is_deepspeed_zeroa_enabled
from .generation.configuration_utils import GenerationConfig
from .trainer import Trainer
from .utils import logging


if TYPE_CHECKING:
						from .data.data_collator import DataCollator
						from .modeling_utils import PreTrainedModel
						from .tokenization_utils_base import PreTrainedTokenizerBase
						from .trainer_callback import TrainerCallback
						from .trainer_utils import EvalPrediction, PredictionOutput
						from .training_args import TrainingArguments


lowerCAmelCase							:						Dict     			=		logging.get_logger(__name__)



class 						__lowercase       (				UpperCAmelCase_			):




			"""simple docstring"""







			def __init__(				self						:  Dict     ,		lowerCAmelCase__						:  Union["PreTrainedModel", nn.Module] = None     ,		lowerCAmelCase__						:  "TrainingArguments" = None     ,		lowerCAmelCase__						:  Optional["DataCollator"] = None     ,		lowerCAmelCase__						:  Optional[Dataset] = None     ,		lowerCAmelCase__						:  Optional[Union[Dataset, Dict[str, Dataset]]] = None     ,		lowerCAmelCase__						:  Optional["PreTrainedTokenizerBase"] = None     ,		lowerCAmelCase__						:  Optional[Callable[[], "PreTrainedModel"]] = None     ,		lowerCAmelCase__						:  Optional[Callable[["EvalPrediction"], Dict]] = None     ,		lowerCAmelCase__						:  Optional[List["TrainerCallback"]] = None     ,		lowerCAmelCase__						:  Tuple[torch.optim.Optimizer, torch.optim.lr_scheduler.LambdaLR] = (None, None)     ,		lowerCAmelCase__						:  Optional[Callable[[torch.Tensor, torch.Tensor], torch.Tensor]] = None     ,		):
									super().__init__(
									    model=lowerCAmelCase__     ,		args=lowerCAmelCase__     ,		data_collator=lowerCAmelCase__     ,		train_dataset=lowerCAmelCase__     ,		eval_dataset=lowerCAmelCase__     ,		tokenizer=lowerCAmelCase__     ,		model_init=lowerCAmelCase__     ,		compute_metrics=lowerCAmelCase__     ,		callbacks=lowerCAmelCase__     ,		optimizers=lowerCAmelCase__     ,		preprocess_logits_for_metrics=lowerCAmelCase__     ,		)

									# Override self.model.generation_config if a GenerationConfig is specified in args.
									# Priority: args.generation_config > model.generation_config > default GenerationConfig.
									if self.args.generation_config is not None:
															SCREAMING_SNAKE_CASE_:				int							= self.load_generation_config(self.args.generation_config)
															SCREAMING_SNAKE_CASE_:				int							= gen_config



			@staticmethod
			def  _SCREAMING_SNAKE_CASE    (				lowerCAmelCase__						:  Union[str, GenerationConfig]):

									# GenerationConfig provided, nothing to do
									if isinstance(lowerCAmelCase__     ,		lowerCAmelCase__):
															return deepcopy(lowerCAmelCase__)

									# str or Path
									SCREAMING_SNAKE_CASE_:				List[str]							= Path(lowerCAmelCase__) if isinstance(lowerCAmelCase__     ,		lowerCAmelCase__) else gen_config_arg
									SCREAMING_SNAKE_CASE_:				Dict							= None

									# Figuring if it is path pointing to a file, pointing to a directory or else a model id or URL
									# This step is required in order to determine config_file_name
									if pretrained_model_name.is_file():
															SCREAMING_SNAKE_CASE_:				Optional[int]							= pretrained_model_name.name
															SCREAMING_SNAKE_CASE_:				Union[str, Any]							= pretrained_model_name.parent
									# dir path
									elif pretrained_model_name.is_dir():
															pass
									# model id or URL
									else:
															SCREAMING_SNAKE_CASE_:				List[Any]							= gen_config_arg

									SCREAMING_SNAKE_CASE_:				Any							= GenerationConfig.from_pretrained(lowerCAmelCase__     ,		lowerCAmelCase__)
									return gen_config



			def  _SCREAMING_SNAKE_CASE    (				self						:  Dict     ,		lowerCAmelCase__						:  Optional[Dataset] = None     ,		lowerCAmelCase__						:  Optional[List[str]] = None     ,		lowerCAmelCase__						:  str = "eval"     ,		**lowerCAmelCase__						:  List[str]     ,		):
									SCREAMING_SNAKE_CASE_:				Optional[int]							= gen_kwargs.copy()
									if gen_kwargs.get("max_length") is None and gen_kwargs.get("max_new_tokens") is None:
															SCREAMING_SNAKE_CASE_:				str							= self.args.generation_max_length
									SCREAMING_SNAKE_CASE_:				Union[str, Any]							= (
									    gen_kwargs["num_beams"] if gen_kwargs.get("num_beams") is not None else self.args.generation_num_beams
									)
									SCREAMING_SNAKE_CASE_:				Any							= gen_kwargs

									return super().evaluate(lowerCAmelCase__     ,		ignore_keys=lowerCAmelCase__     ,		metric_key_prefix=lowerCAmelCase__)



			def  _SCREAMING_SNAKE_CASE    (				self						:  Union[str, Any]     ,		lowerCAmelCase__						:  Dataset     ,		lowerCAmelCase__						:  Optional[List[str]] = None     ,		lowerCAmelCase__						:  str = "test"     ,		**lowerCAmelCase__						:  int     ,		):
									SCREAMING_SNAKE_CASE_:				Optional[Any]							= gen_kwargs.copy()
									if gen_kwargs.get("max_length") is None and gen_kwargs.get("max_new_tokens") is None:
															SCREAMING_SNAKE_CASE_:				Optional[int]							= self.args.generation_max_length
									SCREAMING_SNAKE_CASE_:				Tuple							= (
									    gen_kwargs["num_beams"] if gen_kwargs.get("num_beams") is not None else self.args.generation_num_beams
									)
									SCREAMING_SNAKE_CASE_:				Optional[int]							= gen_kwargs

									return super().predict(lowerCAmelCase__     ,		ignore_keys=lowerCAmelCase__     ,		metric_key_prefix=lowerCAmelCase__)



			def  _SCREAMING_SNAKE_CASE    (				self						:  List[str]     ,		lowerCAmelCase__						:  nn.Module     ,		lowerCAmelCase__						:  Dict[str, Union[torch.Tensor, Any]]     ,		lowerCAmelCase__						:  bool     ,		lowerCAmelCase__						:  Optional[List[str]] = None     ,		**lowerCAmelCase__						:  Dict     ,		):

									if not self.args.predict_with_generate or prediction_loss_only:
															return super().prediction_step(
															    lowerCAmelCase__     ,		lowerCAmelCase__     ,		prediction_loss_only=lowerCAmelCase__     ,		ignore_keys=lowerCAmelCase__)

									SCREAMING_SNAKE_CASE_:				List[Any]							= "labels" in inputs
									SCREAMING_SNAKE_CASE_:				Optional[Any]							= self._prepare_inputs(lowerCAmelCase__)

									# XXX: adapt synced_gpus for fairscale as well
									# Priority (handled in generate):
									# gen_kwargs > model.generation_config > default GenerationConfig()

									if len(lowerCAmelCase__) == 0 and hasattr(self     ,		"_gen_kwargs"):
															SCREAMING_SNAKE_CASE_:				Tuple							= self._gen_kwargs.copy()

									if gen_kwargs.get("max_length") is None and gen_kwargs.get("max_new_tokens") is None:
															SCREAMING_SNAKE_CASE_:				str							= self.model.config.max_length
									SCREAMING_SNAKE_CASE_:				Any							= (
									    gen_kwargs["num_beams"] if gen_kwargs.get("num_beams") is not None else self.model.config.num_beams
									)
									SCREAMING_SNAKE_CASE_:				Optional[Any]							= True if is_deepspeed_zeroa_enabled() else False
									SCREAMING_SNAKE_CASE_:				Optional[int]							= (
									    gen_kwargs["synced_gpus"] if gen_kwargs.get("synced_gpus") is not None else default_synced_gpus
									)

									# If the `decoder_input_ids` was created from `labels`, evict the former, so that the model can freely generate
									# (otherwise, it would continue generating from the padded `decoder_input_ids`)
									if (
									    "labels" in inputs
									    and "decoder_input_ids" in inputs
									    and inputs["labels"].shape == inputs["decoder_input_ids"].shape
									):
															SCREAMING_SNAKE_CASE_:				Optional[int]							= {k: v for k, v in inputs.items() if k != "decoder_input_ids"}
									SCREAMING_SNAKE_CASE_:				int							= self.model.generate(**lowerCAmelCase__     ,		**lowerCAmelCase__)

									# Temporary hack to ensure the generation config is not initialized for each iteration of the evaluation loop
									# TODO: remove this hack when the legacy code that initializes generation_config from a model config is
									# removed in https://github.com/huggingface/transformers/blob/98d88b23f54e5a23e741833f1e973fdf600cc2c5/src/transformers/generation/utils.py#L1183
									if self.model.generation_config._from_model_config:
															SCREAMING_SNAKE_CASE_:				int							= False

									# Retrieves GenerationConfig from model.generation_config
									SCREAMING_SNAKE_CASE_:				Optional[int]							= self.model.generation_config
									# in case the batch is shorter than max length, the output should be padded
									if generated_tokens.shape[-1] < gen_config.max_length:
															SCREAMING_SNAKE_CASE_:				Any							= self._pad_tensors_to_max_len(lowerCAmelCase__     ,		gen_config.max_length)
									elif gen_config.max_new_tokens is not None and generated_tokens.shape[-1] < gen_config.max_new_tokens + 1:
															SCREAMING_SNAKE_CASE_:				Any							= self._pad_tensors_to_max_len(lowerCAmelCase__     ,		gen_config.max_new_tokens + 1)

									with torch.no_grad():
															if has_labels:
																					with self.compute_loss_context_manager():
																											SCREAMING_SNAKE_CASE_:				Union[str, Any]							= model(**lowerCAmelCase__)
																					if self.label_smoother is not None:
																											SCREAMING_SNAKE_CASE_:				int							= self.label_smoother(lowerCAmelCase__     ,		inputs["labels"]).mean().detach()
																					else:
																											SCREAMING_SNAKE_CASE_:				Dict							= (outputs["loss"] if isinstance(lowerCAmelCase__     ,		lowerCAmelCase__) else outputs[0]).mean().detach()
															else:
																					SCREAMING_SNAKE_CASE_:				Optional[int]							= None

									if self.args.prediction_loss_only:
															return loss, None, None

									if has_labels:
															SCREAMING_SNAKE_CASE_:				Union[str, Any]							= inputs["labels"]
															if labels.shape[-1] < gen_config.max_length:
																					SCREAMING_SNAKE_CASE_:				Optional[Any]							= self._pad_tensors_to_max_len(lowerCAmelCase__     ,		gen_config.max_length)
															elif gen_config.max_new_tokens is not None and labels.shape[-1] < gen_config.max_new_tokens + 1:
																					SCREAMING_SNAKE_CASE_:				Optional[int]							= self._pad_tensors_to_max_len(lowerCAmelCase__     ,		gen_config.max_new_tokens + 1)
									else:
															SCREAMING_SNAKE_CASE_:				Any							= None

									return loss, generated_tokens, labels




			def  _SCREAMING_SNAKE_CASE    (				self						:  Union[str, Any]     ,		lowerCAmelCase__						:  Tuple     ,		lowerCAmelCase__						:  List[Any]):
									if self.tokenizer is not None and hasattr(self.tokenizer     ,		"pad_token_id"):
															# If PAD token is not defined at least EOS token has to be defined
															SCREAMING_SNAKE_CASE_:				str							= (
															    self.tokenizer.pad_token_id if self.tokenizer.pad_token_id is not None else self.tokenizer.eos_token_id
															)
									else:
															if self.model.config.pad_token_id is not None:
																					SCREAMING_SNAKE_CASE_:				Tuple							= self.model.config.pad_token_id
															else:
																					raise ValueError("Pad_token_id must be set in the configuration of the model, in order to pad tensors")

									SCREAMING_SNAKE_CASE_:				List[str]							= pad_token_id * torch.ones(
									    (tensor.shape[0], max_length)     ,		dtype=tensor.dtype     ,		device=tensor.device)
									SCREAMING_SNAKE_CASE_:				str							= tensor
									return padded_tensor









from typing import List, Optional, Tuple, Union

import torch

from ...models import UNetaDModel
from ...schedulers import PNDMScheduler
from ...utils import randn_tensor
from ..pipeline_utils import DiffusionPipeline, ImagePipelineOutput



class 						__lowercase       (				UpperCAmelCase_			):




			"""simple docstring"""







			_UpperCAmelCase     :							UNetaDModel
			_UpperCAmelCase     :							PNDMScheduler



			def __init__(				self						:  Optional[int]     ,		lowerCAmelCase__						:  UNetaDModel     ,		lowerCAmelCase__						:  PNDMScheduler):
									super().__init__()

									SCREAMING_SNAKE_CASE_:				Dict							= PNDMScheduler.from_config(scheduler.config)

									self.register_modules(unet=lowerCAmelCase__     ,		scheduler=lowerCAmelCase__)




			@torch.no_grad()
			def __call__(				self						:  int     ,		lowerCAmelCase__						:  int = 1     ,		lowerCAmelCase__						:  int = 50     ,		lowerCAmelCase__						:  Optional[Union[torch.Generator, List[torch.Generator]]] = None     ,		lowerCAmelCase__						:  Optional[str] = "pil"     ,		lowerCAmelCase__						:  bool = True     ,		**lowerCAmelCase__						:  str     ,		):
									SCREAMING_SNAKE_CASE_:				Dict							= randn_tensor(
									    (batch_size, self.unet.config.in_channels, self.unet.config.sample_size, self.unet.config.sample_size)     ,		generator=lowerCAmelCase__     ,		device=self.device     ,		)

									self.scheduler.set_timesteps(lowerCAmelCase__)
									for t in self.progress_bar(self.scheduler.timesteps):
															SCREAMING_SNAKE_CASE_:				Union[str, Any]							= self.unet(lowerCAmelCase__     ,		lowerCAmelCase__).sample

															SCREAMING_SNAKE_CASE_:				Any							= self.scheduler.step(lowerCAmelCase__     ,		lowerCAmelCase__     ,		lowerCAmelCase__).prev_sample

									SCREAMING_SNAKE_CASE_:				Optional[Any]							= (image / 2 + 0.5).clamp(0     ,		1)
									SCREAMING_SNAKE_CASE_:				Dict							= image.cpu().permute(0     ,		2     ,		3     ,		1).numpy()
									if output_type == "pil":
															SCREAMING_SNAKE_CASE_:				Any							= self.numpy_to_pil(lowerCAmelCase__)

									if not return_dict:
															return (image,)

									return ImagePipelineOutput(images=lowerCAmelCase__)






