





'''simple docstring'''







import math
import time
from typing import Dict, List, Optional

from torch.utils.data import Dataset

from transformers import SeqaSeqTrainer, is_torch_tpu_available
from transformers.trainer_utils import PredictionOutput, speed_metrics


if is_torch_tpu_available(check_device=False):
   import torch_xla.core.xla_model as xm
   import torch_xla.debug.metrics as met
class       _lowercase    (							SCREAMING_SNAKE_CASE_       ):







     def __init__(			self     , *A__     , A__=None     , A__=None     , **A__     )			->						Dict:
      super().__init__(*snake_case__     , **snake_case__     )
      snake_case					=      eval_examples
      snake_case					=      post_process_function







     def      UpperCamelCase      (			self     , A__ = None     , A__=None     , A__ = None     , A__ = "eval"     , **A__     , )			->						Optional[Any]:
      snake_case					=      gen_kwargs.copy()
      snake_case					=      (
          gen_kwargs['max_length'] if gen_kwargs.get('''max_length'''     ) is not None else self.args.generation_max_length
      )
      snake_case					=      (
          gen_kwargs['num_beams'] if gen_kwargs.get('''num_beams'''     ) is not None else self.args.generation_num_beams
      )
      snake_case					=      gen_kwargs

      snake_case					=      self.eval_dataset if eval_dataset is None else eval_dataset
      snake_case					=      self.get_eval_dataloader(snake_case__     )
      snake_case					=      self.eval_examples if eval_examples is None else eval_examples

      # Temporarily disable metric computation, we will do it in the loop here.
      snake_case					=      self.compute_metrics
      snake_case					=      None
      snake_case					=      time.time()
      snake_case					=      self.prediction_loop if self.args.use_legacy_prediction_loop else self.evaluation_loop
      try:
       snake_case					=      eval_loop(
           snake_case__     , description='''Evaluation'''     , prediction_loss_only=True if compute_metrics is None else None     , ignore_keys=snake_case__     , metric_key_prefix=snake_case__     , )
      finally:
       snake_case					=      compute_metrics
      snake_case					=      self.args.eval_batch_size * self.args.world_size
      if F"""{metric_key_prefix}_jit_compilation_time""" in output.metrics:
       start_time += output.metrics[F"""{metric_key_prefix}_jit_compilation_time"""]
      output.metrics.update(
          speed_metrics(
              snake_case__     , snake_case__     , num_samples=output.num_samples     , num_steps=math.ceil(output.num_samples / total_batch_size     )     , )     )

      if self.post_process_function is not None and self.compute_metrics is not None and self.args.should_save:
       # Only the main node write the results by default
       snake_case					=      self.post_process_function(snake_case__     , snake_case__     , snake_case__     )
       snake_case					=      self.compute_metrics(snake_case__     )

       # Prefix all keys with metric_key_prefix + '_'
       for key in list(metrics.keys()     ):
        if not key.startswith(F"""{metric_key_prefix}_"""     ):
         snake_case					=      metrics.pop(snake_case__     )

       metrics.update(output.metrics     )
      else:
       snake_case					=      output.metrics

      if self.args.should_log:
       # Only the main node log the results by default
       self.log(snake_case__     )

      if self.args.tpu_metrics_debug or self.args.debug:
       # tpu-comment: Logging debug metrics for PyTorch/XLA (compile, execute times, ops, etc.)
       xm.master_print(met.metrics_report()     )

      snake_case					=      self.callback_handler.on_evaluate(self.args     , self.state     , self.control     , snake_case__     )
      return metrics







     def      UpperCamelCase      (			self     , A__     , A__     , A__=None     , A__ = "test"     , **A__     )			->						Any:
      snake_case					=      gen_kwargs.copy()

      snake_case					=      self.get_test_dataloader(snake_case__     )

      # Temporarily disable metric computation, we will do it in the loop here.
      snake_case					=      self.compute_metrics
      snake_case					=      None
      snake_case					=      time.time()
      snake_case					=      self.prediction_loop if self.args.use_legacy_prediction_loop else self.evaluation_loop
      try:
       snake_case					=      eval_loop(
           snake_case__     , description='''Prediction'''     , prediction_loss_only=True if compute_metrics is None else None     , ignore_keys=snake_case__     , metric_key_prefix=snake_case__     , )
      finally:
       snake_case					=      compute_metrics

      snake_case					=      self.args.eval_batch_size * self.args.world_size
      if F"""{metric_key_prefix}_jit_compilation_time""" in output.metrics:
       start_time += output.metrics[F"""{metric_key_prefix}_jit_compilation_time"""]
      output.metrics.update(
          speed_metrics(
              snake_case__     , snake_case__     , num_samples=output.num_samples     , num_steps=math.ceil(output.num_samples / total_batch_size     )     , )     )
      if self.post_process_function is None or self.compute_metrics is None:
       return output

      snake_case					=      self.post_process_function(snake_case__     , snake_case__     , snake_case__     , '''predict'''     )
      snake_case					=      self.compute_metrics(snake_case__     )

      # Prefix all keys with metric_key_prefix + '_'
      for key in list(metrics.keys()     ):
       if not key.startswith(F"""{metric_key_prefix}_"""     ):
        snake_case					=      metrics.pop(snake_case__     )
      metrics.update(output.metrics     )
      return PredictionOutput(predictions=predictions.predictions     , label_ids=predictions.label_ids     , metrics=snake_case__     )





'''simple docstring'''







import gc
import tempfile
import unittest

import numpy as np
import torch

from diffusers import VersatileDiffusionTextToImagePipeline
from diffusers.utils.testing_utils import nightly, require_torch_gpu, torch_device


lowerCAmelCase     :  str											=    False
class 						UpperCamelCase__       (			unittest.TestCase						):



     """simple docstring"""






     pass
@nightly
@require_torch_gpu
class 						UpperCamelCase__       (			unittest.TestCase						):



     """simple docstring"""






     def a						(			self ):

            '''simple docstring'''

            super().tearDown()
            gc.collect()
            torch.cuda.empty_cache()







     def a						(			self ):

            '''simple docstring'''

            _lowerCAmelCase     :				List[str]       =							VersatileDiffusionTextToImagePipeline.from_pretrained('shi-labs/versatile-diffusion' )
            # remove text_unet
            pipe.remove_unused_weights()
            pipe.to(snake_case__ )
            pipe.set_progress_bar_config(disable=snake_case__ )

            _lowerCAmelCase     :				Any       =							'A painting of a squirrel eating a burger '
            _lowerCAmelCase     :				List[Any]       =							torch.manual_seed(0 )
            _lowerCAmelCase     :				Any       =							pipe(
                prompt=snake_case__   ,						generator=snake_case__   ,						guidance_scale=7.5   ,						num_inference_steps=2   ,						output_type='numpy' ).images

            with tempfile.TemporaryDirectory() as tmpdirname:
                   pipe.save_pretrained(snake_case__ )
                   _lowerCAmelCase     :				Any       =							VersatileDiffusionTextToImagePipeline.from_pretrained(snake_case__ )
            pipe.to(snake_case__ )
            pipe.set_progress_bar_config(disable=snake_case__ )

            _lowerCAmelCase     :				Optional[int]       =							generator.manual_seed(0 )
            _lowerCAmelCase     :				Tuple       =							pipe(
                prompt=snake_case__   ,						generator=snake_case__   ,						guidance_scale=7.5   ,						num_inference_steps=2   ,						output_type='numpy' ).images

            assert np.abs(image - new_image ).sum() < 1E-5, "Models don't have the same forward pass"







     def a						(			self ):

            '''simple docstring'''

            _lowerCAmelCase     :				Union[str, Any]       =							VersatileDiffusionTextToImagePipeline.from_pretrained(
                'shi-labs/versatile-diffusion'   ,						torch_dtype=torch.floataa )
            pipe.to(snake_case__ )
            pipe.set_progress_bar_config(disable=snake_case__ )

            _lowerCAmelCase     :				Dict       =							'A painting of a squirrel eating a burger '
            _lowerCAmelCase     :				Any       =							torch.manual_seed(0 )
            _lowerCAmelCase     :				int       =							pipe(
                prompt=snake_case__   ,						generator=snake_case__   ,						guidance_scale=7.5   ,						num_inference_steps=50   ,						output_type='numpy' ).images

            _lowerCAmelCase     :				Optional[int]       =							image[0, 253:256, 253:256, -1]

            assert image.shape == (1, 512, 512, 3)
            _lowerCAmelCase     :				Dict       =							np.array([0.3367, 0.3169, 0.2656, 0.3870, 0.4790, 0.3796, 0.4009, 0.4878, 0.4778] )

            assert np.abs(image_slice.flatten() - expected_slice ).max() < 1E-2

