






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_(		lowerCAmelCase__			,							lowerCAmelCase__			,							lowerCAmelCase__			,							unittest.TestCase	):



     """simple docstring"""



     __snake_case		:				Optional[int]  					=StableDiffusionModelEditingPipeline
     __snake_case		:				List[Any]  					=TEXT_TO_IMAGE_PARAMS
     __snake_case		:				Optional[Any]  					=TEXT_TO_IMAGE_BATCH_PARAMS
     __snake_case		:				str  					=TEXT_TO_IMAGE_IMAGE_PARAMS
     __snake_case		:				Any  					=TEXT_TO_IMAGE_IMAGE_PARAMS





     def     __UpperCamelCase						(		self					:	Optional[int])							->	Tuple:

      """simple docstring"""
      torch.manual_seed(0)
      SCREAMING_SNAKE_CASE				    =		UNetaDConditionModel(
          block_out_channels=(3_2, 6_4)  ,  layers_per_block=2  ,  sample_size=3_2  ,  in_channels=4  ,  out_channels=4  ,  down_block_types=('DownBlock2D', 'CrossAttnDownBlock2D')  ,  up_block_types=('CrossAttnUpBlock2D', 'UpBlock2D')  ,  cross_attention_dim=3_2  ,  )
      SCREAMING_SNAKE_CASE				    =		DDIMScheduler()
      torch.manual_seed(0)
      SCREAMING_SNAKE_CASE				    =		AutoencoderKL(
          block_out_channels=[3_2, 6_4]  ,  in_channels=3  ,  out_channels=3  ,  down_block_types=['DownEncoderBlock2D', 'DownEncoderBlock2D']  ,  up_block_types=['UpDecoderBlock2D', 'UpDecoderBlock2D']  ,  latent_channels=4  ,  )
      torch.manual_seed(0)
      SCREAMING_SNAKE_CASE				    =		CLIPTextConfig(
          bos_token_id=0  ,  eos_token_id=2  ,  hidden_size=3_2  ,  intermediate_size=3_7  ,  layer_norm_eps=1e-05  ,  num_attention_heads=4  ,  num_hidden_layers=5  ,  pad_token_id=1  ,  vocab_size=1_0_0_0  ,  )
      SCREAMING_SNAKE_CASE				    =		CLIPTextModel(_UpperCAmelCase)
      SCREAMING_SNAKE_CASE				    =		CLIPTokenizer.from_pretrained('hf-internal-testing/tiny-random-clip')

      SCREAMING_SNAKE_CASE				    =		{
          'unet': unet,
          'scheduler': scheduler,
          'vae': vae,
          'text_encoder': text_encoder,
          'tokenizer': tokenizer,
          'safety_checker': None,
          'feature_extractor': None,
      }
      return components





     def     __UpperCamelCase						(		self					:	Optional[Any]  ,  lowerCAmelCase__					:	List[str]  ,  lowerCAmelCase__					:	Dict=0)							->	Optional[Any]:

      """simple docstring"""
      SCREAMING_SNAKE_CASE				    =		torch.manual_seed(_UpperCAmelCase)
      SCREAMING_SNAKE_CASE				    =		{
          '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     __UpperCamelCase						(		self					:	Dict)							->	List[str]:

      """simple docstring"""
      SCREAMING_SNAKE_CASE				    =		'cpu'  # ensure determinism for the device-dependent torch.Generator
      SCREAMING_SNAKE_CASE				    =		self.get_dummy_components()
      SCREAMING_SNAKE_CASE				    =		StableDiffusionModelEditingPipeline(**_UpperCAmelCase)
      SCREAMING_SNAKE_CASE				    =		sd_pipe.to(_UpperCAmelCase)
      sd_pipe.set_progress_bar_config(disable=_UpperCAmelCase)

      SCREAMING_SNAKE_CASE				    =		self.get_dummy_inputs(_UpperCAmelCase)
      SCREAMING_SNAKE_CASE				    =		sd_pipe(**_UpperCAmelCase).images
      SCREAMING_SNAKE_CASE				    =		image[0, -3:, -3:, -1]
      assert image.shape == (1, 6_4, 6_4, 3)

      SCREAMING_SNAKE_CASE				    =		np.array([0.47_55, 0.51_32, 0.49_76, 0.39_04, 0.35_54, 0.47_65, 0.51_39, 0.51_58, 0.48_89])

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





     def     __UpperCamelCase						(		self					:	int)							->	List[str]:

      """simple docstring"""
      SCREAMING_SNAKE_CASE				    =		'cpu'  # ensure determinism for the device-dependent torch.Generator
      SCREAMING_SNAKE_CASE				    =		self.get_dummy_components()
      SCREAMING_SNAKE_CASE				    =		StableDiffusionModelEditingPipeline(**_UpperCAmelCase)
      SCREAMING_SNAKE_CASE				    =		sd_pipe.to(_UpperCAmelCase)
      sd_pipe.set_progress_bar_config(disable=_UpperCAmelCase)

      SCREAMING_SNAKE_CASE				    =		self.get_dummy_inputs(_UpperCAmelCase)
      SCREAMING_SNAKE_CASE				    =		'french fries'
      SCREAMING_SNAKE_CASE				    =		sd_pipe(**_UpperCAmelCase  ,  negative_prompt=_UpperCAmelCase)
      SCREAMING_SNAKE_CASE				    =		output.images
      SCREAMING_SNAKE_CASE				    =		image[0, -3:, -3:, -1]

      assert image.shape == (1, 6_4, 6_4, 3)

      SCREAMING_SNAKE_CASE				    =		np.array([0.49_92, 0.51_01, 0.50_04, 0.39_49, 0.36_04, 0.47_35, 0.52_16, 0.52_04, 0.49_13])

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





     def     __UpperCamelCase						(		self					:	Optional[Any])							->	List[str]:

      """simple docstring"""
      SCREAMING_SNAKE_CASE				    =		'cpu'  # ensure determinism for the device-dependent torch.Generator
      SCREAMING_SNAKE_CASE				    =		self.get_dummy_components()
      SCREAMING_SNAKE_CASE				    =		EulerAncestralDiscreteScheduler(
          beta_start=0.0_00_85  ,  beta_end=0.0_12  ,  beta_schedule='scaled_linear')
      SCREAMING_SNAKE_CASE				    =		StableDiffusionModelEditingPipeline(**_UpperCAmelCase)
      SCREAMING_SNAKE_CASE				    =		sd_pipe.to(_UpperCAmelCase)
      sd_pipe.set_progress_bar_config(disable=_UpperCAmelCase)

      SCREAMING_SNAKE_CASE				    =		self.get_dummy_inputs(_UpperCAmelCase)
      SCREAMING_SNAKE_CASE				    =		sd_pipe(**_UpperCAmelCase).images
      SCREAMING_SNAKE_CASE				    =		image[0, -3:, -3:, -1]

      assert image.shape == (1, 6_4, 6_4, 3)

      SCREAMING_SNAKE_CASE				    =		np.array([0.47_47, 0.53_72, 0.47_79, 0.49_82, 0.55_43, 0.48_16, 0.52_38, 0.49_04, 0.50_27])

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





     def     __UpperCamelCase						(		self					:	Optional[Any])							->	Optional[int]:

      """simple docstring"""
      SCREAMING_SNAKE_CASE				    =		'cpu'  # ensure determinism for the device-dependent torch.Generator
      SCREAMING_SNAKE_CASE				    =		self.get_dummy_components()
      SCREAMING_SNAKE_CASE				    =		PNDMScheduler()
      SCREAMING_SNAKE_CASE				    =		StableDiffusionModelEditingPipeline(**_UpperCAmelCase)
      SCREAMING_SNAKE_CASE				    =		sd_pipe.to(_UpperCAmelCase)
      sd_pipe.set_progress_bar_config(disable=_UpperCAmelCase)

      SCREAMING_SNAKE_CASE				    =		self.get_dummy_inputs(_UpperCAmelCase)
      # the pipeline does not expect pndm so test if it raises error.
      with self.assertRaises(_UpperCAmelCase):
       SCREAMING_SNAKE_CASE				    =		sd_pipe(**_UpperCAmelCase).images





     def     __UpperCamelCase						(		self					:	Optional[int])							->	List[str]:

      """simple docstring"""
      super().test_inference_batch_single_identical(expected_max_diff=5e-3)





     def     __UpperCamelCase						(		self					:	Tuple)							->	Dict:

      """simple docstring"""
      super().test_attention_slicing_forward_pass(expected_max_diff=5e-3)

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



     """simple docstring"""



     def     __UpperCamelCase						(		self					:	List[Any])							->	Optional[int]:

      """simple docstring"""
      super().tearDown()
      gc.collect()
      torch.cuda.empty_cache()





     def     __UpperCamelCase						(		self					:	Optional[int]  ,  lowerCAmelCase__					:	Optional[int]=0)							->	Optional[Any]:

      """simple docstring"""
      SCREAMING_SNAKE_CASE				    =		torch.manual_seed(_UpperCAmelCase)
      SCREAMING_SNAKE_CASE				    =		{
          'prompt': 'A field of roses',
          'generator': generator,
          'num_inference_steps': 3,
          'guidance_scale': 7.5,
          'output_type': 'numpy',
      }
      return inputs





     def     __UpperCamelCase						(		self					:	Optional[Any])							->	Dict:

      """simple docstring"""
      SCREAMING_SNAKE_CASE				    =		'CompVis/stable-diffusion-v1-4'
      SCREAMING_SNAKE_CASE				    =		StableDiffusionModelEditingPipeline.from_pretrained(_UpperCAmelCase  ,  safety_checker=_UpperCAmelCase)
      pipe.to(_UpperCAmelCase)
      pipe.set_progress_bar_config(disable=_UpperCAmelCase)
      pipe.enable_attention_slicing()

      SCREAMING_SNAKE_CASE				    =		self.get_inputs()
      SCREAMING_SNAKE_CASE				    =		pipe(**_UpperCAmelCase).images
      SCREAMING_SNAKE_CASE				    =		image[0, -3:, -3:, -1].flatten()

      assert image.shape == (1, 5_1_2, 5_1_2, 3)

      SCREAMING_SNAKE_CASE				    =		np.array(
          [0.6_74_94_96, 0.6_38_64_53, 0.51_44_32_67, 0.66_09_49_05, 0.61_92_12_15, 0.5_49_13_32, 0.5_74_44_17, 0.58_07_51_06, 0.5_17_46_58])

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

      SCREAMING_SNAKE_CASE				    =		pipe(**_UpperCAmelCase).images
      SCREAMING_SNAKE_CASE				    =		image[0, -3:, -3:, -1].flatten()

      assert image.shape == (1, 5_1_2, 5_1_2, 3)

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





     def     __UpperCamelCase						(		self					:	List[Any])							->	List[Any]:

      """simple docstring"""
      torch.cuda.empty_cache()
      torch.cuda.reset_max_memory_allocated()
      torch.cuda.reset_peak_memory_stats()

      SCREAMING_SNAKE_CASE				    =		'CompVis/stable-diffusion-v1-4'
      SCREAMING_SNAKE_CASE				    =		DDIMScheduler.from_pretrained(_UpperCAmelCase  ,  subfolder='scheduler')
      SCREAMING_SNAKE_CASE				    =		StableDiffusionModelEditingPipeline.from_pretrained(
          _UpperCAmelCase  ,  scheduler=_UpperCAmelCase  ,  safety_checker=_UpperCAmelCase)
      SCREAMING_SNAKE_CASE				    =		pipe.to(_UpperCAmelCase)
      pipe.set_progress_bar_config(disable=_UpperCAmelCase)
      pipe.enable_attention_slicing(1)
      pipe.enable_sequential_cpu_offload()

      SCREAMING_SNAKE_CASE				    =		self.get_inputs()
      SCREAMING_SNAKE_CASE				    =		pipe(**_UpperCAmelCase)

      SCREAMING_SNAKE_CASE				    =		torch.cuda.max_memory_allocated()
      # make sure that less than 4.4 GB is allocated
      assert mem_bytes < 4.4 * 1_0**9






import logging
from pathlib import Path

import numpy as np
import pytorch_lightning as pl
import torch
from pytorch_lightning.callbacks import EarlyStopping, ModelCheckpoint
from pytorch_lightning.utilities import rank_zero_only

from utils import save_json




def 				__SCREAMING_SNAKE_CASE  (						SCREAMING_SNAKE_CASE			: List[Any]     )	->		str:
				__lowercase								=   filter(lambda SCREAMING_SNAKE_CASE     : p.requires_grad       ,				model.parameters()     )
				__lowercase								=   sum([np.prod(p.size()     ) for p in model_parameters]     )
				return params


SCREAMING_SNAKE_CASE__				=							logging.getLogger(__name__)
class       A__  (		pl.Callback							):





							def      a__					(					self    :  List[str]	,					_UpperCAmelCase    :  Tuple	,					_UpperCAmelCase    :  List[str]							)		->						Tuple:



											"""simple docstring"""


											__lowercase								=   {f"""lr_group_{i}""": param['lr'] for i, param in enumerate(pl_module.trainer.optimizers[0].param_groups							)}
											pl_module.logger.log_metrics(_UpperCAmelCase							)





							@rank_zero_only
							def      a__					(					self    :  Dict	,					_UpperCAmelCase    :  pl.Trainer	,					_UpperCAmelCase    :  pl.LightningModule	,					_UpperCAmelCase    :  str	,					_UpperCAmelCase    :  int=True							)		->						None:



											"""simple docstring"""


											logger.info(f"""***** {type_path} results at step {trainer.global_step:05d} *****"""							)
											__lowercase								=   trainer.callback_metrics
											trainer.logger.log_metrics({k: v for k, v in metrics.items() if k not in ['log', 'progress_bar', 'preds']}							)
											# Log results
											__lowercase								=   Path(pl_module.hparams.output_dir							)
											if type_path == "test":
															__lowercase								=   od / 'test_results.txt'
															__lowercase								=   od / 'test_generations.txt'
											else:
															# this never gets hit. I prefer not to save intermediate generations, and results are in metrics.json
															# If people want this it will be easy enough to add back.
															__lowercase								=   od / f"""{type_path}_results/{trainer.global_step:05d}.txt"""
															__lowercase								=   od / f"""{type_path}_generations/{trainer.global_step:05d}.txt"""
															results_file.parent.mkdir(exist_ok=_UpperCAmelCase							)
															generations_file.parent.mkdir(exist_ok=_UpperCAmelCase							)
											with open(_UpperCAmelCase	,					'a+'							) as writer:
															for key in sorted(_UpperCAmelCase							):
																			if key in ["log", "progress_bar", "preds"]:
																							continue
																			__lowercase								=   metrics[key]
																			if isinstance(_UpperCAmelCase	,					torch.Tensor							):
																							__lowercase								=   val.item()
																			__lowercase								=   f"""{key}: {val:.6f}\n"""
																			writer.write(_UpperCAmelCase							)

											if not save_generations:
															return

											if "preds" in metrics:
															__lowercase								=   '\n'.join(metrics['preds']							)
															generations_file.open('w+'							).write(_UpperCAmelCase							)





							@rank_zero_only
							def      a__					(					self    :  str	,					_UpperCAmelCase    :  Optional[int]	,					_UpperCAmelCase    :  str							)		->						Optional[int]:



											"""simple docstring"""


											try:
															__lowercase								=   pl_module.model.model.num_parameters()
											except AttributeError:
															__lowercase								=   pl_module.model.num_parameters()

											__lowercase								=   count_trainable_parameters(_UpperCAmelCase							)
											# mp stands for million parameters
											trainer.logger.log_metrics({'n_params': npars, 'mp': npars / 1e6, 'grad_mp': n_trainable_pars / 1e6}							)





							@rank_zero_only
							def      a__					(					self    :  Union[str, Any]	,					_UpperCAmelCase    :  pl.Trainer	,					_UpperCAmelCase    :  pl.LightningModule							)		->						Optional[Any]:



											"""simple docstring"""


											save_json(pl_module.metrics	,					pl_module.metrics_save_path							)
											return self._write_logs(_UpperCAmelCase	,					_UpperCAmelCase	,					'test'							)





							@rank_zero_only
							def      a__					(					self    :  int	,					_UpperCAmelCase    :  pl.Trainer	,					_UpperCAmelCase    :  Tuple							)		->						str:



											"""simple docstring"""


											save_json(pl_module.metrics	,					pl_module.metrics_save_path							)
											# Uncommenting this will save val generations
											# return self._write_logs(trainer, pl_module, "valid")






def 				__SCREAMING_SNAKE_CASE  (						SCREAMING_SNAKE_CASE			: List[Any]       ,				SCREAMING_SNAKE_CASE			: List[str]       ,				SCREAMING_SNAKE_CASE			: Union[str, Any]=1       ,				SCREAMING_SNAKE_CASE			: str=False     )	->		List[str]:
				if metric == "rouge2":
								__lowercase								=   '{val_avg_rouge2:.4f}-{step_count}'
				elif metric == "bleu":
								__lowercase								=   '{val_avg_bleu:.4f}-{step_count}'
				elif metric == "loss":
								__lowercase								=   '{val_avg_loss:.4f}-{step_count}'
				else:
								raise NotImplementedError(
								    F"""seq2seq callbacks only support rouge2, bleu and loss, got {metric}, You can make your own by adding to"""
								    ' this function.'     )

				__lowercase								=   ModelCheckpoint(
				    dirpath=SCREAMING_SNAKE_CASE       ,				filename=SCREAMING_SNAKE_CASE       ,				monitor=F"""val_{metric}"""       ,				mode='min' if 'loss' in metric else 'max'       ,				save_top_k=SCREAMING_SNAKE_CASE       ,				)
				return checkpoint_callback




def 				__SCREAMING_SNAKE_CASE  (						SCREAMING_SNAKE_CASE			: Any       ,				SCREAMING_SNAKE_CASE			: int     )	->		Union[str, Any]:
				return EarlyStopping(
				    monitor=F"""val_{metric}"""       ,				mode='min' if 'loss' in metric else 'max'       ,				patience=SCREAMING_SNAKE_CASE       ,				verbose=SCREAMING_SNAKE_CASE       ,				)





