code: 
  
  
  
  
  """simple docstring"""
  
  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
  
  
  __A 		=			logging.get_logger(__name__)
  class 			lowerCamelCase__    (  lowerCamelCase_			):
  
  
  			def __init__(     self				, SCREAMING_SNAKE_CASE = None				, SCREAMING_SNAKE_CASE = None				, SCREAMING_SNAKE_CASE = None				, SCREAMING_SNAKE_CASE = None				, SCREAMING_SNAKE_CASE = None				, SCREAMING_SNAKE_CASE = None				, SCREAMING_SNAKE_CASE = None				, SCREAMING_SNAKE_CASE = None				, SCREAMING_SNAKE_CASE = None				, SCREAMING_SNAKE_CASE = (None, None)				, SCREAMING_SNAKE_CASE = None				, ):
  
  
  
  					"""simple docstring"""
  
  
  
  
  
  
  
  					super().__init__(
  					    model=SCREAMING_SNAKE_CASE				, args=SCREAMING_SNAKE_CASE				, data_collator=SCREAMING_SNAKE_CASE				, train_dataset=SCREAMING_SNAKE_CASE				, eval_dataset=SCREAMING_SNAKE_CASE				, tokenizer=SCREAMING_SNAKE_CASE				, model_init=SCREAMING_SNAKE_CASE				, compute_metrics=SCREAMING_SNAKE_CASE				, callbacks=SCREAMING_SNAKE_CASE				, optimizers=SCREAMING_SNAKE_CASE				, preprocess_logits_for_metrics=SCREAMING_SNAKE_CASE				, )
  
  					# 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:
  							snake_case		:						Any								=				self.load_generation_config(self.args.generation_config							)
  							snake_case		:						List[Any]								=				gen_config
  
  
  			@staticmethod
  			def 							lowerCamelCase_				(     SCREAMING_SNAKE_CASE							):
  
  
  
  					"""simple docstring"""
  
  
  
  
  
  
  
  					if isinstance(SCREAMING_SNAKE_CASE				, SCREAMING_SNAKE_CASE							):
  							return deepcopy(SCREAMING_SNAKE_CASE							)
  
  					# str or Path
  					snake_case		:						Dict								=				Path(SCREAMING_SNAKE_CASE							) if isinstance(SCREAMING_SNAKE_CASE				, SCREAMING_SNAKE_CASE							) else gen_config_arg
  					snake_case		:						Tuple								=				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():
  							snake_case		:						List[Any]								=				pretrained_model_name.name
  							snake_case		:						int								=				pretrained_model_name.parent
  					# dir path
  					elif pretrained_model_name.is_dir():
  							pass
  					# model id or URL
  					else:
  							snake_case		:						List[Any]								=				gen_config_arg
  
  					snake_case		:						Tuple								=				GenerationConfig.from_pretrained(SCREAMING_SNAKE_CASE				, SCREAMING_SNAKE_CASE							)
  					return gen_config
  
  
  			def 							lowerCamelCase_				(     self				, SCREAMING_SNAKE_CASE = None				, SCREAMING_SNAKE_CASE = None				, SCREAMING_SNAKE_CASE = "eval"				, **SCREAMING_SNAKE_CASE				, ):
  
  
  
  					"""simple docstring"""
  
  
  
  
  
  
  
  					snake_case		:						Tuple								=				gen_kwargs.copy()
  					if gen_kwargs.get("max_length"							) is None and gen_kwargs.get("max_new_tokens"							) is None:
  							snake_case		:						Optional[Any]								=				self.args.generation_max_length
  					snake_case		:						str								=				(
  					    gen_kwargs["num_beams"] if gen_kwargs.get("num_beams"							) is not None else self.args.generation_num_beams
  					)
  					snake_case		:						Dict								=				gen_kwargs
  
  					return super().evaluate(SCREAMING_SNAKE_CASE				, ignore_keys=SCREAMING_SNAKE_CASE				, metric_key_prefix=SCREAMING_SNAKE_CASE							)
  
  
  			def 							lowerCamelCase_				(     self				, SCREAMING_SNAKE_CASE				, SCREAMING_SNAKE_CASE = None				, SCREAMING_SNAKE_CASE = "test"				, **SCREAMING_SNAKE_CASE				, ):
  
  
  
  					"""simple docstring"""
  
  
  
  
  
  
  
  					snake_case		:						Dict								=				gen_kwargs.copy()
  					if gen_kwargs.get("max_length"							) is None and gen_kwargs.get("max_new_tokens"							) is None:
  							snake_case		:						Tuple								=				self.args.generation_max_length
  					snake_case		:						str								=				(
  					    gen_kwargs["num_beams"] if gen_kwargs.get("num_beams"							) is not None else self.args.generation_num_beams
  					)
  					snake_case		:						Tuple								=				gen_kwargs
  
  					return super().predict(SCREAMING_SNAKE_CASE				, ignore_keys=SCREAMING_SNAKE_CASE				, metric_key_prefix=SCREAMING_SNAKE_CASE							)
  
  
  			def 							lowerCamelCase_				(     self				, SCREAMING_SNAKE_CASE				, SCREAMING_SNAKE_CASE				, SCREAMING_SNAKE_CASE				, SCREAMING_SNAKE_CASE = None				, **SCREAMING_SNAKE_CASE				, ):
  
  
  
  					"""simple docstring"""
  
  
  
  
  
  
  
  					if not self.args.predict_with_generate or prediction_loss_only:
  							return super().prediction_step(
  							    SCREAMING_SNAKE_CASE				, SCREAMING_SNAKE_CASE				, prediction_loss_only=SCREAMING_SNAKE_CASE				, ignore_keys=SCREAMING_SNAKE_CASE							)
  
  					snake_case		:						Any								=				"labels" in inputs
  					snake_case		:						Dict								=				self._prepare_inputs(SCREAMING_SNAKE_CASE							)
  
  					# XXX: adapt synced_gpus for fairscale as well
  					# Priority (handled in generate):
  					# gen_kwargs > model.generation_config > default GenerationConfig()
  
  					if len(SCREAMING_SNAKE_CASE							) == 0 and hasattr(self				, "_gen_kwargs"							):
  							snake_case		:						Optional[Any]								=				self._gen_kwargs.copy()
  
  					if gen_kwargs.get("max_length"							) is None and gen_kwargs.get("max_new_tokens"							) is None:
  							snake_case		:						str								=				self.model.config.max_length
  					snake_case		:						Optional[int]								=				(
  					    gen_kwargs["num_beams"] if gen_kwargs.get("num_beams"							) is not None else self.model.config.num_beams
  					)
  					snake_case		:						Union[str, Any]								=				True if is_deepspeed_zeroa_enabled() else False
  					snake_case		:						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
  					):
  							snake_case		:						Optional[int]								=				{k: v for k, v in inputs.items() if k != "decoder_input_ids"}
  					snake_case		:						List[Any]								=				self.model.generate(**SCREAMING_SNAKE_CASE				, **SCREAMING_SNAKE_CASE							)
  
  					# 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:
  							snake_case		:						str								=				False
  
  					# Retrieves GenerationConfig from model.generation_config
  					snake_case		:						Optional[Any]								=				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:
  							snake_case		:						str								=				self._pad_tensors_to_max_len(SCREAMING_SNAKE_CASE				, gen_config.max_length							)
  					elif gen_config.max_new_tokens is not None and generated_tokens.shape[-1] < gen_config.max_new_tokens + 1:
  							snake_case		:						str								=				self._pad_tensors_to_max_len(SCREAMING_SNAKE_CASE				, gen_config.max_new_tokens + 1							)
  
  					with torch.no_grad():
  							if has_labels:
  									with self.compute_loss_context_manager():
  											snake_case		:						Optional[Any]								=				model(**SCREAMING_SNAKE_CASE							)
  									if self.label_smoother is not None:
  											snake_case		:						Union[str, Any]								=				self.label_smoother(SCREAMING_SNAKE_CASE				, inputs["labels"]							).mean().detach()
  									else:
  											snake_case		:						Tuple								=				(outputs["loss"] if isinstance(SCREAMING_SNAKE_CASE				, SCREAMING_SNAKE_CASE							) else outputs[0]).mean().detach()
  							else:
  									snake_case		:						Dict								=				None
  
  					if self.args.prediction_loss_only:
  							return loss, None, None
  
  					if has_labels:
  							snake_case		:						List[str]								=				inputs["labels"]
  							if labels.shape[-1] < gen_config.max_length:
  									snake_case		:						int								=				self._pad_tensors_to_max_len(SCREAMING_SNAKE_CASE				, gen_config.max_length							)
  							elif gen_config.max_new_tokens is not None and labels.shape[-1] < gen_config.max_new_tokens + 1:
  									snake_case		:						Optional[Any]								=				self._pad_tensors_to_max_len(SCREAMING_SNAKE_CASE				, gen_config.max_new_tokens + 1							)
  					else:
  							snake_case		:						Union[str, Any]								=				None
  
  					return loss, generated_tokens, labels
  
  
  
  			def 							lowerCamelCase_				(     self				, SCREAMING_SNAKE_CASE				, SCREAMING_SNAKE_CASE							):
  
  
  
  					"""simple docstring"""
  
  
  
  
  
  
  
  					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
  							snake_case		:						List[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:
  									snake_case		:						Any								=				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"							)
  
  					snake_case		:						List[str]								=				pad_token_id * torch.ones(
  					    (tensor.shape[0], max_length)				, dtype=tensor.dtype				, device=tensor.device							)
  					snake_case		:						Union[str, Any]								=				tensor
  					return padded_tensor
  
  
style_context: 
  
  
  
  
  """simple docstring"""
  
  import unittest
  
  import numpy as np
  import PIL
  import torch
  
  from diffusers.image_processor import VaeImageProcessor
  class 			lowerCamelCase__    (  unittest.TestCase			):
  
  
  			@property
  			def 							lowerCamelCase_				(     self							):
  
  
  
  					"""simple docstring"""
  
  
  
  
  
  
  
  					snake_case		:						int								=				1
  					snake_case		:						Dict								=				3
  					snake_case		:						int								=				8
  					snake_case		:						int								=				8
  
  					snake_case		:						List[str]								=				torch.rand((batch_size, num_channels, height, width)							)
  
  					return sample
  
  
  			def 							lowerCamelCase_				(     self				, SCREAMING_SNAKE_CASE							):
  
  
  
  					"""simple docstring"""
  
  
  
  
  
  
  
  					if isinstance(image[0]				, PIL.Image.Image							):
  							return np.stack([np.array(SCREAMING_SNAKE_CASE							) for i in image]				, axis=0							)
  					elif isinstance(SCREAMING_SNAKE_CASE				, torch.Tensor							):
  							return image.cpu().numpy().transpose(0				, 2				, 3				, 1							)
  					return image
  
  
  			def 							lowerCamelCase_				(     self							):
  
  
  
  					"""simple docstring"""
  
  
  
  
  
  
  
  					snake_case		:						Optional[int]								=				VaeImageProcessor(do_resize=SCREAMING_SNAKE_CASE				, do_normalize=SCREAMING_SNAKE_CASE							)
  
  					snake_case		:						Any								=				self.dummy_sample
  					snake_case		:						Optional[Any]								=				self.to_np(SCREAMING_SNAKE_CASE							)
  
  					for output_type in ["pt", "np", "pil"]:
  							snake_case		:						Optional[Any]								=				image_processor.postprocess(
  							    image_processor.preprocess(SCREAMING_SNAKE_CASE							)				, output_type=SCREAMING_SNAKE_CASE				, )
  							snake_case		:						Optional[Any]								=				self.to_np(SCREAMING_SNAKE_CASE							)
  							snake_case		:						Any								=				(input_np * 255).round() if output_type == "pil" else input_np
  							assert (
  							    np.abs(in_np - out_np							).max() < 1E-6
  							), F'''decoded output does not match input for output_type {output_type}'''
  
  
  			def 							lowerCamelCase_				(     self							):
  
  
  
  					"""simple docstring"""
  
  
  
  
  
  
  
  					snake_case		:						Dict								=				VaeImageProcessor(do_resize=SCREAMING_SNAKE_CASE				, do_normalize=SCREAMING_SNAKE_CASE							)
  					snake_case		:						Tuple								=				self.dummy_sample.cpu().numpy().transpose(0				, 2				, 3				, 1							)
  
  					for output_type in ["pt", "np", "pil"]:
  							snake_case		:						Dict								=				image_processor.postprocess(image_processor.preprocess(SCREAMING_SNAKE_CASE							)				, output_type=SCREAMING_SNAKE_CASE							)
  
  							snake_case		:						Union[str, Any]								=				self.to_np(SCREAMING_SNAKE_CASE							)
  							snake_case		:						Tuple								=				(input_np * 255).round() if output_type == "pil" else input_np
  							assert (
  							    np.abs(in_np - out_np							).max() < 1E-6
  							), F'''decoded output does not match input for output_type {output_type}'''
  
  
  			def 							lowerCamelCase_				(     self							):
  
  
  
  					"""simple docstring"""
  
  
  
  
  
  
  
  					snake_case		:						Optional[Any]								=				VaeImageProcessor(do_resize=SCREAMING_SNAKE_CASE				, do_normalize=SCREAMING_SNAKE_CASE							)
  
  					snake_case		:						str								=				self.dummy_sample.cpu().numpy().transpose(0				, 2				, 3				, 1							)
  					snake_case		:						int								=				image_processor.numpy_to_pil(SCREAMING_SNAKE_CASE							)
  
  					for output_type in ["pt", "np", "pil"]:
  							snake_case		:						Dict								=				image_processor.postprocess(image_processor.preprocess(SCREAMING_SNAKE_CASE							)				, output_type=SCREAMING_SNAKE_CASE							)
  							for i, o in zip(SCREAMING_SNAKE_CASE				, SCREAMING_SNAKE_CASE							):
  									snake_case		:						Dict								=				np.array(SCREAMING_SNAKE_CASE							)
  									snake_case		:						Tuple								=				self.to_np(SCREAMING_SNAKE_CASE							) if output_type == "pil" else (self.to_np(SCREAMING_SNAKE_CASE							) * 255).round()
  									assert (
  									    np.abs(in_np - out_np							).max() < 1E-6
  									), F'''decoded output does not match input for output_type {output_type}'''
  
  
  			def 							lowerCamelCase_				(     self							):
  
  
  
  					"""simple docstring"""
  
  
  
  
  
  
  
  					snake_case		:						Tuple								=				VaeImageProcessor(do_resize=SCREAMING_SNAKE_CASE				, do_normalize=SCREAMING_SNAKE_CASE							)
  
  					snake_case		:						List[str]								=				self.dummy_sample
  					snake_case		:						Optional[int]								=				input_pt_ad.squeeze(0							)
  
  					snake_case		:						Any								=				image_processor.postprocess(
  					    image_processor.preprocess(SCREAMING_SNAKE_CASE							)				, output_type="np"				, )
  					snake_case		:						Union[str, Any]								=				image_processor.postprocess(
  					    image_processor.preprocess(SCREAMING_SNAKE_CASE							)				, output_type="np"				, )
  
  					snake_case		:						Union[str, Any]								=				self.to_np(self.dummy_sample							)
  					snake_case		:						Tuple								=				input_np_ad.squeeze(0							)
  
  					snake_case		:						str								=				image_processor.postprocess(
  					    image_processor.preprocess(SCREAMING_SNAKE_CASE							)				, output_type="np"				, )
  					snake_case		:						Union[str, Any]								=				image_processor.postprocess(
  					    image_processor.preprocess(SCREAMING_SNAKE_CASE							)				, output_type="np"				, )
  
  					assert np.abs(out_pt_ad - out_pt_ad							).max() < 1E-6
  					assert np.abs(out_np_ad - out_np_ad							).max() < 1E-6
  
  
  
  			def 							lowerCamelCase_				(     self							):
  
  
  
  					"""simple docstring"""
  
  
  
  
  
  
  
  					snake_case		:						Optional[Any]								=				VaeImageProcessor(do_resize=SCREAMING_SNAKE_CASE				, do_normalize=SCREAMING_SNAKE_CASE							)
  
  					snake_case		:						Union[str, Any]								=				self.dummy_sample
  					snake_case		:						List[str]								=				list(SCREAMING_SNAKE_CASE							)
  
  					snake_case		:						str								=				image_processor.postprocess(
  					    image_processor.preprocess(SCREAMING_SNAKE_CASE							)				, output_type="np"				, )
  
  					snake_case		:						Any								=				image_processor.postprocess(
  					    image_processor.preprocess(SCREAMING_SNAKE_CASE							)				, output_type="np"				, )
  
  					snake_case		:						Dict								=				self.to_np(self.dummy_sample							)
  					list(SCREAMING_SNAKE_CASE							)
  
  					snake_case		:						str								=				image_processor.postprocess(
  					    image_processor.preprocess(SCREAMING_SNAKE_CASE							)				, output_type="np"				, )
  
  					snake_case		:						Union[str, Any]								=				image_processor.postprocess(
  					    image_processor.preprocess(SCREAMING_SNAKE_CASE							)				, output_type="np"				, )
  
  					assert np.abs(out_pt_ad - out_pt_list							).max() < 1E-6
  					assert np.abs(out_np_ad - out_np_list							).max() < 1E-6
  
  