




'''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      UpperCamelCase       (	snake_case__      ,		snake_case__      ,		snake_case__      ,		unittest.TestCase       ):
  a     :Any          =					StableDiffusionModelEditingPipeline
  a     :str          =					TEXT_TO_IMAGE_PARAMS
  a     :int          =					TEXT_TO_IMAGE_BATCH_PARAMS
  a     :Optional[int]          =					TEXT_TO_IMAGE_IMAGE_PARAMS
  a     :int          =					TEXT_TO_IMAGE_IMAGE_PARAMS



  def     a_							(			self ):







      '''simple docstring'''


      torch.manual_seed(0 )
      _a				:							Dict							       =      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		,     )
      _a				:							Tuple							       =      DDIMScheduler()
      torch.manual_seed(0 )
      _a				:							str							       =      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 )
      _a				:							Tuple							       =      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=1000		,     )
      _a				:							Any							       =      CLIPTextModel(__lowerCAmelCase )
      _a				:							str							       =      CLIPTokenizer.from_pretrained("""hf-internal-testing/tiny-random-clip""" )

      _a				:							Union[str, Any]							       =      {
          """unet""": unet,
          """scheduler""": scheduler,
          """vae""": vae,
          """text_encoder""": text_encoder,
          """tokenizer""": tokenizer,
          """safety_checker""": None,
          """feature_extractor""": None,
      }
      return components



  def     a_							(			self		,     __lowerCAmelCase		,     __lowerCAmelCase=0 ):







      '''simple docstring'''


      _a				:							Optional[Any]							       =      torch.manual_seed(__lowerCAmelCase )
      _a				:							str							       =      {
          """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     a_							(			self ):







      '''simple docstring'''


      _a				:							Tuple							       =      """cpu"""  # ensure determinism for the device-dependent torch.Generator
      _a				:							List[str]							       =      self.get_dummy_components()
      _a				:							List[str]							       =      StableDiffusionModelEditingPipeline(**__lowerCAmelCase )
      _a				:							Any							       =      sd_pipe.to(__lowerCAmelCase )
      sd_pipe.set_progress_bar_config(disable=__lowerCAmelCase )

      _a				:							Dict							       =      self.get_dummy_inputs(__lowerCAmelCase )
      _a				:							List[Any]							       =      sd_pipe(**__lowerCAmelCase ).images
      _a				:							Optional[Any]							       =      image[0, -3:, -3:, -1]
      assert image.shape == (1, 64, 64, 3)

      _a				:							int							       =      np.array([0.4755, 0.5132, 0.4976, 0.3904, 0.3554, 0.4765, 0.5139, 0.5158, 0.4889] )

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



  def     a_							(			self ):







      '''simple docstring'''


      _a				:							Tuple							       =      """cpu"""  # ensure determinism for the device-dependent torch.Generator
      _a				:							Dict							       =      self.get_dummy_components()
      _a				:							int							       =      StableDiffusionModelEditingPipeline(**__lowerCAmelCase )
      _a				:							Optional[Any]							       =      sd_pipe.to(__lowerCAmelCase )
      sd_pipe.set_progress_bar_config(disable=__lowerCAmelCase )

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

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

      _a				:							Dict							       =      np.array([0.4992, 0.5101, 0.5004, 0.3949, 0.3604, 0.4735, 0.5216, 0.5204, 0.4913] )

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



  def     a_							(			self ):







      '''simple docstring'''


      _a				:							Optional[Any]							       =      """cpu"""  # ensure determinism for the device-dependent torch.Generator
      _a				:							Tuple							       =      self.get_dummy_components()
      _a				:							int							       =      EulerAncestralDiscreteScheduler(
          beta_start=0.00085		,     beta_end=0.012		,     beta_schedule="""scaled_linear""" )
      _a				:							List[Any]							       =      StableDiffusionModelEditingPipeline(**__lowerCAmelCase )
      _a				:							List[str]							       =      sd_pipe.to(__lowerCAmelCase )
      sd_pipe.set_progress_bar_config(disable=__lowerCAmelCase )

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

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

      _a				:							Dict							       =      np.array([0.4747, 0.5372, 0.4779, 0.4982, 0.5543, 0.4816, 0.5238, 0.4904, 0.5027] )

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



  def     a_							(			self ):







      '''simple docstring'''


      _a				:							Tuple							       =      """cpu"""  # ensure determinism for the device-dependent torch.Generator
      _a				:							Any							       =      self.get_dummy_components()
      _a				:							Union[str, Any]							       =      PNDMScheduler()
      _a				:							List[str]							       =      StableDiffusionModelEditingPipeline(**__lowerCAmelCase )
      _a				:							List[Any]							       =      sd_pipe.to(__lowerCAmelCase )
      sd_pipe.set_progress_bar_config(disable=__lowerCAmelCase )

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



  def     a_							(			self ):







      '''simple docstring'''


      super().test_inference_batch_single_identical(expected_max_diff=5e-3 )



  def     a_							(			self ):







      '''simple docstring'''


      super().test_attention_slicing_forward_pass(expected_max_diff=5e-3 )






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



  def     a_							(			self ):







      '''simple docstring'''


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



  def     a_							(			self		,     __lowerCAmelCase=0 ):







      '''simple docstring'''


      _a				:							Optional[Any]							       =      torch.manual_seed(__lowerCAmelCase )
      _a				:							List[str]							       =      {
          """prompt""": """A field of roses""",
          """generator""": generator,
          """num_inference_steps""": 3,
          """guidance_scale""": 7.5,
          """output_type""": """numpy""",
      }
      return inputs



  def     a_							(			self ):







      '''simple docstring'''


      _a				:							Union[str, Any]							       =      """CompVis/stable-diffusion-v1-4"""
      _a				:							Tuple							       =      StableDiffusionModelEditingPipeline.from_pretrained(__lowerCAmelCase		,     safety_checker=__lowerCAmelCase )
      pipe.to(__lowerCAmelCase )
      pipe.set_progress_bar_config(disable=__lowerCAmelCase )
      pipe.enable_attention_slicing()

      _a				:							List[str]							       =      self.get_inputs()
      _a				:							str							       =      pipe(**__lowerCAmelCase ).images
      _a				:							str							       =      image[0, -3:, -3:, -1].flatten()

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

      _a				:							Tuple							       =      np.array(
          [0.6749496, 0.6386453, 0.51443267, 0.66094905, 0.61921215, 0.5491332, 0.5744417, 0.58075106, 0.5174658] )

      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""" )

      _a				:							Union[str, Any]							       =      pipe(**__lowerCAmelCase ).images
      _a				:							Tuple							       =      image[0, -3:, -3:, -1].flatten()

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

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



  def     a_							(			self ):







      '''simple docstring'''


      torch.cuda.empty_cache()
      torch.cuda.reset_max_memory_allocated()
      torch.cuda.reset_peak_memory_stats()

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

      _a				:							Dict							       =      self.get_inputs()
      _a				:							Tuple							       =      pipe(**__lowerCAmelCase )

      _a				:							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'''





import functools
from contextlib import contextmanager

import bert_score
from packaging import version

import datasets





@contextmanager
def UpperCAmelCase    (       ):
    def filter_log(UpperCAmelCase_						):
        return False if "This IS expected if you are initializing" in record.msg else True

    _a				:							Optional[Any]							       =      datasets.utils.logging.get_logger("""transformers.modeling_utils"""						)
    logger.addFilter(UpperCAmelCase_						)
    try:
        yield
    finally:
        logger.removeFilter(UpperCAmelCase_						)


a		:	Tuple       =      '''\
@inproceedings{bert-score,
  title={BERTScore: Evaluating Text Generation with BERT},
  author={Tianyi Zhang* and Varsha Kishore* and Felix Wu* and Kilian Q. Weinberger and Yoav Artzi},
  booktitle={International Conference on Learning Representations},
  year={2020},
  url={https://openreview.net/forum?id=SkeHuCVFDr}
}
'''

a		:	List[str]       =      '''\
BERTScore leverages the pre-trained contextual embeddings from BERT and matches words in candidate and reference
sentences by cosine similarity.
It has been shown to correlate with human judgment on sentence-level and system-level evaluation.
Moreover, BERTScore computes precision, recall, and F1 measure, which can be useful for evaluating different language
generation tasks.

See the project\'s README at https://github.com/Tiiiger/bert_score#readme for more information.
'''

a		:	Dict       =      '''
BERTScore Metrics with the hashcode from a source against one or more references.

Args:
    predictions (list of str): Prediction/candidate sentences.
    references (list of str or list of list of str): Reference sentences.
    lang (str): Language of the sentences; required (e.g. \'en\').
    model_type (str): Bert specification, default using the suggested
        model for the target language; has to specify at least one of
        `model_type` or `lang`.
    num_layers (int): The layer of representation to use,
        default using the number of layers tuned on WMT16 correlation data.
    verbose (bool): Turn on intermediate status update.
    idf (bool or dict): Use idf weighting; can also be a precomputed idf_dict.
    device (str): On which the contextual embedding model will be allocated on.
        If this argument is None, the model lives on cuda:0 if cuda is available.
    nthreads (int): Number of threads.
    batch_size (int): Bert score processing batch size,
        at least one of `model_type` or `lang`. `lang` needs to be
        specified when `rescale_with_baseline` is True.
    rescale_with_baseline (bool): Rescale bertscore with pre-computed baseline.
    baseline_path (str): Customized baseline file.
    use_fast_tokenizer (bool): `use_fast` parameter passed to HF tokenizer. New in version 0.3.10.

Returns:
    precision: Precision.
    recall: Recall.
    f1: F1 score.
    hashcode: Hashcode of the library.

Examples:

    >>> predictions = ["hello there", "general kenobi"]
    >>> references = ["hello there", "general kenobi"]
    >>> bertscore = datasets.load_metric("bertscore")
    >>> results = bertscore.compute(predictions=predictions, references=references, lang="en")
    >>> print([round(v, 2) for v in results["f1"]])
    [1.0, 1.0]
'''





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



  def     a_							(			self ):







      '''simple docstring'''


      return datasets.MetricInfo(
          description=_DESCRIPTION		,     citation=_CITATION		,     homepage="""https://github.com/Tiiiger/bert_score"""		,     inputs_description=_KWARGS_DESCRIPTION		,     features=datasets.Features(
              {
                  """predictions""": datasets.Value("""string"""		,     id="""sequence""" ),
                  """references""": datasets.Sequence(datasets.Value("""string"""		,     id="""sequence""" )		,     id="""references""" ),
              } )		,     codebase_urls=["""https://github.com/Tiiiger/bert_score"""]		,     reference_urls=[
              """https://github.com/Tiiiger/bert_score""",
              """https://arxiv.org/abs/1904.09675""",
          ]		,     )



  def     a_							(			self		,     __lowerCAmelCase		,     __lowerCAmelCase		,     __lowerCAmelCase=None		,     __lowerCAmelCase=None		,     __lowerCAmelCase=None		,     __lowerCAmelCase=False		,     __lowerCAmelCase=False		,     __lowerCAmelCase=None		,     __lowerCAmelCase=64		,     __lowerCAmelCase=4		,     __lowerCAmelCase=False		,     __lowerCAmelCase=False		,     __lowerCAmelCase=None		,     __lowerCAmelCase=False		,     ):







      '''simple docstring'''


      _a				:							Any							       =      bert_score.utils.get_hash
      _a				:							int							       =      bert_score.BERTScorer

      if version.parse(bert_score.__version__ ) >= version.parse("""0.3.10""" ):
          _a				:							Union[str, Any]							       =      functools.partial(__lowerCAmelCase		,     use_fast_tokenizer=__lowerCAmelCase )
          _a				:							Dict							       =      functools.partial(__lowerCAmelCase		,     use_fast_tokenizer=__lowerCAmelCase )
      elif use_fast_tokenizer:
          raise ImportWarning(
              """To use a fast tokenizer, the module `bert-score>=0.3.10` is required, and the current version of `bert-score` doesn't match this condition.\n"""
              """You can install it with `pip install \"bert-score>=0.3.10\"`.""" )

      if model_type is None:
          assert lang is not None, "either lang or model_type should be specified"
          _a				:							List[str]							       =      bert_score.utils.langamodel[lang.lower()]

      if num_layers is None:
          _a				:							Dict							       =      bert_score.utils.modelalayers[model_type]

      _a				:							Union[str, Any]							       =      get_hash(
          model=__lowerCAmelCase		,     num_layers=__lowerCAmelCase		,     idf=__lowerCAmelCase		,     rescale_with_baseline=__lowerCAmelCase		,     use_custom_baseline=baseline_path is not None		,     )

      with filter_logging_context():
          if not hasattr(self		,     """cached_bertscorer""" ) or self.cached_bertscorer.hash != hashcode:
              _a				:							Optional[Any]							       =      scorer(
                  model_type=__lowerCAmelCase		,     num_layers=__lowerCAmelCase		,     batch_size=__lowerCAmelCase		,     nthreads=__lowerCAmelCase		,     all_layers=__lowerCAmelCase		,     idf=__lowerCAmelCase		,     device=__lowerCAmelCase		,     lang=__lowerCAmelCase		,     rescale_with_baseline=__lowerCAmelCase		,     baseline_path=__lowerCAmelCase		,     )

      ((_a)       , (_a)       , (_a))				:							Optional[Any]							       =      self.cached_bertscorer.score(
          cands=__lowerCAmelCase		,     refs=__lowerCAmelCase		,     verbose=__lowerCAmelCase		,     batch_size=__lowerCAmelCase		,     )
      _a				:							Tuple							       =      {
          """precision""": P.tolist(),
          """recall""": R.tolist(),
          """f1""": F.tolist(),
          """hashcode""": hashcode,
      }
      return output_dict



  def     a_							(			self		,     __lowerCAmelCase=None		,     __lowerCAmelCase=None		,     **__lowerCAmelCase ):







      '''simple docstring'''


      if references is not None:
          _a				:							Dict							       =      [[ref] if isinstance(__lowerCAmelCase		,     __lowerCAmelCase ) else ref for ref in references]
      super().add_batch(predictions=__lowerCAmelCase		,     references=__lowerCAmelCase		,     **__lowerCAmelCase )



  def     a_							(			self		,     __lowerCAmelCase=None		,     __lowerCAmelCase=None		,     **__lowerCAmelCase ):







      '''simple docstring'''


      if isinstance(__lowerCAmelCase		,     __lowerCAmelCase ):
          _a				:							Optional[Any]							       =      [reference]
      super().add(prediction=__lowerCAmelCase		,     reference=__lowerCAmelCase		,     **__lowerCAmelCase )


