




'''simple docstring'''







import gc
import unittest

import numpy as np
import torch
from transformers import CLIPTextConfig, CLIPTextModel, CLIPTokenizer

from diffusers import (
    AutoencoderKL,
    DDIMScheduler,
    EulerAncestralDiscreteScheduler,
    PNDMScheduler,
    StableDiffusionModelEditingPipeline,
    UNetaDConditionModel,
)
from diffusers.utils import slow, torch_device
from diffusers.utils.testing_utils import enable_full_determinism, require_torch_gpu, skip_mps

from ..pipeline_params import TEXT_TO_IMAGE_BATCH_PARAMS, TEXT_TO_IMAGE_IMAGE_PARAMS, TEXT_TO_IMAGE_PARAMS
from ..test_pipelines_common import PipelineKarrasSchedulerTesterMixin, PipelineLatentTesterMixin, PipelineTesterMixin


enable_full_determinism()







@skip_mps
class 				_a   (a__,		a__,		a__,		unittest.TestCase  ):





     '''simple docstring'''



     lowerCAmelCase_     :    int									=				StableDiffusionModelEditingPipeline
     lowerCAmelCase_     :    Dict									=				TEXT_TO_IMAGE_PARAMS
     lowerCAmelCase_     :    Optional[int]									=				TEXT_TO_IMAGE_BATCH_PARAMS
     lowerCAmelCase_     :    Tuple									=				TEXT_TO_IMAGE_IMAGE_PARAMS
     lowerCAmelCase_     :    List[str]									=				TEXT_TO_IMAGE_IMAGE_PARAMS







     def  snake_case_			(							self       )			->							str:
          torch.manual_seed(0       )
          snake_case :			Tuple					   = UNetaDConditionModel(
              block_out_channels=(32, 64)       ,layers_per_block=2       ,sample_size=32       ,in_channels=4       ,out_channels=4       ,down_block_types=("""DownBlock2D""", """CrossAttnDownBlock2D""")       ,up_block_types=("""CrossAttnUpBlock2D""", """UpBlock2D""")       ,cross_attention_dim=32       ,)
          snake_case :			List[str]					   = DDIMScheduler()
          torch.manual_seed(0       )
          snake_case :			List[Any]					   = AutoencoderKL(
              block_out_channels=[32, 64]       ,in_channels=3       ,out_channels=3       ,down_block_types=["""DownEncoderBlock2D""", """DownEncoderBlock2D"""]       ,up_block_types=["""UpDecoderBlock2D""", """UpDecoderBlock2D"""]       ,latent_channels=4       ,)
          torch.manual_seed(0       )
          snake_case :			List[Any]					   = CLIPTextConfig(
              bos_token_id=0       ,eos_token_id=2       ,hidden_size=32       ,intermediate_size=37       ,layer_norm_eps=1E-05       ,num_attention_heads=4       ,num_hidden_layers=5       ,pad_token_id=1       ,vocab_size=1_000       ,)
          snake_case :			str					   = CLIPTextModel(__a       )
          snake_case :			List[Any]					   = CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip"""       )

          snake_case :			Tuple					   = {
              """unet""": unet,
              """scheduler""": scheduler,
              """vae""": vae,
              """text_encoder""": text_encoder,
              """tokenizer""": tokenizer,
              """safety_checker""": None,
              """feature_extractor""": None,
          }
          return components







     def  snake_case_			(							self       ,__a       ,__a=0       )			->							Any:
          snake_case :			int					   = torch.manual_seed(__a       )
          snake_case :			Any					   = {
              """prompt""": """A field of roses""",
              """generator""": generator,
              # Setting height and width to None to prevent OOMs on CPU.
              """height""": None,
              """width""": None,
              """num_inference_steps""": 2,
              """guidance_scale""": 6.0,
              """output_type""": """numpy""",
          }
          return inputs







     def  snake_case_			(							self       )			->							List[Any]:
          snake_case :			List[str]					   = """cpu"""  # ensure determinism for the device-dependent torch.Generator
          snake_case :			int					   = self.get_dummy_components()
          snake_case :			Dict					   = StableDiffusionModelEditingPipeline(**__a       )
          snake_case :			Optional[int]					   = sd_pipe.to(__a       )
          sd_pipe.set_progress_bar_config(disable=__a       )

          snake_case :			Dict					   = self.get_dummy_inputs(__a       )
          snake_case :			Any					   = sd_pipe(**__a       ).images
          snake_case :			str					   = image[0, -3:, -3:, -1]
          assert image.shape == (1, 64, 64, 3)

          snake_case :			Optional[Any]					   = np.array([0.4_755, 0.5_132, 0.4_976, 0.3_904, 0.3_554, 0.4_765, 0.5_139, 0.5_158, 0.4_889]       )

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







     def  snake_case_			(							self       )			->							List[Any]:
          snake_case :			Optional[int]					   = """cpu"""  # ensure determinism for the device-dependent torch.Generator
          snake_case :			str					   = self.get_dummy_components()
          snake_case :			Optional[Any]					   = StableDiffusionModelEditingPipeline(**__a       )
          snake_case :			Dict					   = sd_pipe.to(__a       )
          sd_pipe.set_progress_bar_config(disable=__a       )

          snake_case :			Dict					   = self.get_dummy_inputs(__a       )
          snake_case :			Any					   = """french fries"""
          snake_case :			List[str]					   = sd_pipe(**__a       ,negative_prompt=__a       )
          snake_case :			Tuple					   = output.images
          snake_case :			Optional[Any]					   = image[0, -3:, -3:, -1]

          assert image.shape == (1, 64, 64, 3)

          snake_case :			List[Any]					   = np.array([0.4_992, 0.5_101, 0.5_004, 0.3_949, 0.3_604, 0.4_735, 0.5_216, 0.5_204, 0.4_913]       )

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







     def  snake_case_			(							self       )			->							Optional[int]:
          snake_case :			int					   = """cpu"""  # ensure determinism for the device-dependent torch.Generator
          snake_case :			int					   = self.get_dummy_components()
          snake_case :			int					   = EulerAncestralDiscreteScheduler(
              beta_start=0.00_085       ,beta_end=0.012       ,beta_schedule="""scaled_linear"""       )
          snake_case :			Dict					   = StableDiffusionModelEditingPipeline(**__a       )
          snake_case :			Dict					   = sd_pipe.to(__a       )
          sd_pipe.set_progress_bar_config(disable=__a       )

          snake_case :			Dict					   = self.get_dummy_inputs(__a       )
          snake_case :			str					   = sd_pipe(**__a       ).images
          snake_case :			List[str]					   = image[0, -3:, -3:, -1]

          assert image.shape == (1, 64, 64, 3)

          snake_case :			Union[str, Any]					   = np.array([0.4_747, 0.5_372, 0.4_779, 0.4_982, 0.5_543, 0.4_816, 0.5_238, 0.4_904, 0.5_027]       )

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







     def  snake_case_			(							self       )			->							List[Any]:
          snake_case :			str					   = """cpu"""  # ensure determinism for the device-dependent torch.Generator
          snake_case :			List[Any]					   = self.get_dummy_components()
          snake_case :			int					   = PNDMScheduler()
          snake_case :			List[Any]					   = StableDiffusionModelEditingPipeline(**__a       )
          snake_case :			str					   = sd_pipe.to(__a       )
          sd_pipe.set_progress_bar_config(disable=__a       )

          snake_case :			List[Any]					   = self.get_dummy_inputs(__a       )
          # the pipeline does not expect pndm so test if it raises error.
          with self.assertRaises(__a       ):
               snake_case :			List[Any]					   = sd_pipe(**__a       ).images







     def  snake_case_			(							self       )			->							List[Any]:
          super().test_inference_batch_single_identical(expected_max_diff=5E-3       )







     def  snake_case_			(							self       )			->							Tuple:
          super().test_attention_slicing_forward_pass(expected_max_diff=5E-3       )







@slow
@require_torch_gpu
class 				_a   (unittest.TestCase  ):





     '''simple docstring'''



     def  snake_case_			(							self       )			->							Any:
          super().tearDown()
          gc.collect()
          torch.cuda.empty_cache()







     def  snake_case_			(							self       ,__a=0       )			->							Tuple:
          snake_case :			int					   = torch.manual_seed(__a       )
          snake_case :			int					   = {
              """prompt""": """A field of roses""",
              """generator""": generator,
              """num_inference_steps""": 3,
              """guidance_scale""": 7.5,
              """output_type""": """numpy""",
          }
          return inputs







     def  snake_case_			(							self       )			->							Any:
          snake_case :			List[Any]					   = """CompVis/stable-diffusion-v1-4"""
          snake_case :			List[Any]					   = StableDiffusionModelEditingPipeline.from_pretrained(__a       ,safety_checker=__a       )
          pipe.to(__a       )
          pipe.set_progress_bar_config(disable=__a       )
          pipe.enable_attention_slicing()

          snake_case :			List[Any]					   = self.get_inputs()
          snake_case :			Optional[int]					   = pipe(**__a       ).images
          snake_case :			str					   = image[0, -3:, -3:, -1].flatten()

          assert image.shape == (1, 512, 512, 3)

          snake_case :			Optional[Any]					   = np.array(
              [0.6_749_496, 0.6_386_453, 0.51_443_267, 0.66_094_905, 0.61_921_215, 0.5_491_332, 0.5_744_417, 0.58_075_106, 0.5_174_658]       )

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

          # make sure image changes after editing
          pipe.edit_model("""A pack of roses"""       ,"""A pack of blue roses"""       )

          snake_case :			int					   = pipe(**__a       ).images
          snake_case :			Union[str, Any]					   = image[0, -3:, -3:, -1].flatten()

          assert image.shape == (1, 512, 512, 3)

          assert np.abs(expected_slice - image_slice       ).max() > 1E-1







     def  snake_case_			(							self       )			->							List[Any]:
          torch.cuda.empty_cache()
          torch.cuda.reset_max_memory_allocated()
          torch.cuda.reset_peak_memory_stats()

          snake_case :			int					   = """CompVis/stable-diffusion-v1-4"""
          snake_case :			Union[str, Any]					   = DDIMScheduler.from_pretrained(__a       ,subfolder="""scheduler"""       )
          snake_case :			int					   = StableDiffusionModelEditingPipeline.from_pretrained(
              __a       ,scheduler=__a       ,safety_checker=__a       )
          snake_case :			Union[str, Any]					   = pipe.to(__a       )
          pipe.set_progress_bar_config(disable=__a       )
          pipe.enable_attention_slicing(1       )
          pipe.enable_sequential_cpu_offload()

          snake_case :			Optional[int]					   = self.get_inputs()
          snake_case :			int					   = pipe(**__a       )

          snake_case :			Union[str, Any]					   = torch.cuda.max_memory_allocated()
          # make sure that less than 4.4 GB is allocated
          assert mem_bytes < 4.4 * 10**9









'''simple docstring'''







from sklearn.metrics import mean_absolute_error

import datasets


lowercase			:      int			=					"""\
@article{scikit-learn,
 title={Scikit-learn: Machine Learning in {P}ython},
 author={Pedregosa, F. and Varoquaux, G. and Gramfort, A. and Michel, V.
         and Thirion, B. and Grisel, O. and Blondel, M. and Prettenhofer, P.
         and Weiss, R. and Dubourg, V. and Vanderplas, J. and Passos, A. and
         Cournapeau, D. and Brucher, M. and Perrot, M. and Duchesnay, E.},
 journal={Journal of Machine Learning Research},
 volume={12},
 pages={2825--2830},
 year={2011}
}
"""

lowercase			:      Optional[Any]			=					"""\
Mean Absolute Error (MAE) is the mean of the magnitude of difference between the predicted and actual
values.
"""


lowercase			:      List[Any]			=					"""
Args:
    predictions: array-like of shape (n_samples,) or (n_samples, n_outputs)
        Estimated target values.
    references: array-like of shape (n_samples,) or (n_samples, n_outputs)
        Ground truth (correct) target values.
    sample_weight: array-like of shape (n_samples,), default=None
        Sample weights.
    multioutput: {\"raw_values\", \"uniform_average\"} or array-like of shape (n_outputs,), default=\"uniform_average\"
        Defines aggregating of multiple output values. Array-like value defines weights used to average errors.

                 \"raw_values\" : Returns a full set of errors in case of multioutput input.

                 \"uniform_average\" : Errors of all outputs are averaged with uniform weight.

Returns:
    mae : mean absolute error.
        If multioutput is \"raw_values\", then mean absolute error is returned for each output separately. If multioutput is \"uniform_average\" or an ndarray of weights, then the weighted average of all output errors is returned.
        MAE output is non-negative floating point. The best value is 0.0.
Examples:

    >>> mae_metric = datasets.load_metric(\"mae\")
    >>> predictions = [2.5, 0.0, 2, 8]
    >>> references = [3, -0.5, 2, 7]
    >>> results = mae_metric.compute(predictions=predictions, references=references)
    >>> print(results)
    {'mae': 0.5}

    If you're using multi-dimensional lists, then set the config as follows :

    >>> mae_metric = datasets.load_metric(\"mae\", \"multilist\")
    >>> predictions = [[0.5, 1], [-1, 1], [7, -6]]
    >>> references = [[0, 2], [-1, 2], [8, -5]]
    >>> results = mae_metric.compute(predictions=predictions, references=references)
    >>> print(results)
    {'mae': 0.75}
    >>> results = mae_metric.compute(predictions=predictions, references=references, multioutput='raw_values')
    >>> print(results)
    {'mae': array([0.5, 1. ])}
"""







@datasets.utils.file_utils.add_start_docstrings(_DESCRIPTION,		_KWARGS_DESCRIPTION  )
class 				_a   (datasets.Metric  ):





     '''simple docstring'''



     def  snake_case_			(							self       )			->							int:
          return datasets.MetricInfo(
              description=_DESCRIPTION       ,citation=_CITATION       ,inputs_description=_KWARGS_DESCRIPTION       ,features=datasets.Features(self._get_feature_types()       )       ,reference_urls=[
                  """https://scikit-learn.org/stable/modules/generated/sklearn.metrics.mean_absolute_error.html"""
              ]       ,)







     def  snake_case_			(							self       )			->							Any:
          if self.config_name == "multilist":
               return {
                   "predictions": datasets.Sequence(datasets.Value("""float"""       )       ),
                   "references": datasets.Sequence(datasets.Value("""float"""       )       ),
               }
          else:
               return {
                   "predictions": datasets.Value("""float"""       ),
                   "references": datasets.Value("""float"""       ),
               }







     def  snake_case_			(							self       ,__a       ,__a       ,__a=None       ,__a="uniform_average"       )			->							str:
          snake_case :			int					   = mean_absolute_error(__a       ,__a       ,sample_weight=__a       ,multioutput=__a       )

          return {"mae": mae_score}



