





import random
import unittest

import torch

from diffusers import IFImgaImgSuperResolutionPipeline
from diffusers.utils import floats_tensor
from diffusers.utils.import_utils import is_xformers_available
from diffusers.utils.testing_utils import skip_mps, torch_device

from ..pipeline_params import TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS, TEXT_GUIDED_IMAGE_VARIATION_PARAMS
from ..test_pipelines_common import PipelineTesterMixin
from . import IFPipelineTesterMixin

@skip_mps
class   lowerCAmelCase				(   __UpperCamelCase,    __UpperCamelCase,    unittest.TestCase     ):
 UpperCAmelCase__						=						IFImgaImgSuperResolutionPipeline
 UpperCAmelCase__						=						TEXT_GUIDED_IMAGE_VARIATION_PARAMS - {"""width""", """height"""}
 UpperCAmelCase__						=						TEXT_GUIDED_IMAGE_VARIATION_BATCH_PARAMS.union({"""original_image"""}     )
 UpperCAmelCase__						=						PipelineTesterMixin.required_optional_params - {"""latents"""}
 def 	A_      ( self    : int      )     ->      str:
     return self._get_superresolution_dummy_components()
 def 	A_      ( self    : Any     ,	UpperCAmelCase    : Tuple     ,	UpperCAmelCase    : Dict=0      )     ->      List[Any]:
     if str(UpperCAmelCase      ).startswith('mps'      ):
         lowerCamelCase__       : Union[str, Any]   					=						torch.manual_seed(UpperCAmelCase      )
     else:
         lowerCamelCase__       : Optional[int]   					=						torch.Generator(device=UpperCAmelCase      ).manual_seed(UpperCAmelCase      )

     lowerCamelCase__       : Union[str, Any]   					=						floats_tensor((1, 3, 32, 32)     ,	rng=random.Random(UpperCAmelCase      )      ).to(UpperCAmelCase      )
     lowerCamelCase__       : Union[str, Any]   					=						floats_tensor((1, 3, 16, 16)     ,	rng=random.Random(UpperCAmelCase      )      ).to(UpperCAmelCase      )

     lowerCamelCase__       : Union[str, Any]   					=						{
         'prompt': 'A painting of a squirrel eating a burger',
         'image': image,
         'original_image': original_image,
         'generator': generator,
         'num_inference_steps': 2,
         'output_type': 'numpy',
     }

     return inputs
 @unittest.skipIf(
     torch_device != 'cuda' or not is_xformers_available()     ,	reason='XFormers attention is only available with CUDA and `xformers` installed'     ,	)
 def 	A_      ( self    : str      )     ->      str:
     self._test_xformers_attention_forwardGenerator_pass(expected_max_diff=1e-3      )
 def 	A_      ( self    : List[Any]      )     ->      List[Any]:
     self._test_save_load_optional_components()
 @unittest.skipIf(torch_device != 'cuda'     ,	reason='float16 requires CUDA'      )
 def 	A_      ( self    : Union[str, Any]      )     ->      Tuple:
     # Due to non-determinism in save load of the hf-internal-testing/tiny-random-t5 text encoder
     super().test_save_load_floataa(expected_max_diff=1e-1      )
 def 	A_      ( self    : List[Any]      )     ->      str:
     self._test_attention_slicing_forward_pass(expected_max_diff=1e-2      )
 def 	A_      ( self    : str      )     ->      int:
     self._test_save_load_local()







 def 	A_      ( self    : Optional[int]      )     ->      str:
     self._test_inference_batch_single_identical(
         expected_max_diff=1e-2     ,	)












import gc
import threading
import time

import psutil
import torch

class   lowerCAmelCase				:
 def __init__( self    : str      )     ->      Union[str, Any]:
     lowerCamelCase__       : Optional[Any]   					=						psutil.Process()
     lowerCamelCase__       : Union[str, Any]   					=						False
 def 	A_      ( self    : Optional[int]      )     ->      int:
     lowerCamelCase__       : Optional[Any]   					=						-1

     while True:
         lowerCamelCase__       : Dict   					=						max(self.process.memory_info().rss     ,	self.cpu_memory_peak      )

         # can't sleep or will not catch the peak right (this comment is here on purpose)
         if not self.peak_monitoring:
             break
 def 	A_      ( self    : Tuple      )     ->      Dict:
     lowerCamelCase__       : List[Any]   					=						True
     lowerCamelCase__       : List[str]   					=						threading.Thread(target=self.peak_monitor      )
     lowerCamelCase__       : Union[str, Any]   					=						True
     self.thread.start()







 def 	A_      ( self    : str      )     ->      Dict:
     lowerCamelCase__       : int   					=						False
     self.thread.join()
     return self.cpu_memory_peak


_UpperCAmelCase							:			Dict        =							PeakCPUMemory()


def 			SCREAMING_SNAKE_CASE (			)     ->	Union[str, Any]:
    # Time
    lowerCamelCase__       : List[Any]   					=						{'time': time.time()}

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

    # CPU mem
    lowerCamelCase__       : List[str]   					=						psutil.Process().memory_info().rss
    cpu_peak_tracker.start()

    # GPU mem
    for i in range(torch.cuda.device_count()					):
        lowerCamelCase__       : Union[str, Any]   					=						torch.cuda.memory_allocated(_UpperCAmelCase					)
    torch.cuda.reset_peak_memory_stats()

    return measures


def 			SCREAMING_SNAKE_CASE (			_UpperCAmelCase					)     ->	Tuple:
    # Time
    lowerCamelCase__       : Optional[int]   					=						{'time': time.time() - start_measures['time']}

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

    # CPU mem
    lowerCamelCase__       : Dict   					=						(psutil.Process().memory_info().rss - start_measures['cpu']) / 2**20
    lowerCamelCase__       : int   					=						(cpu_peak_tracker.stop() - start_measures['cpu']) / 2**20

    # GPU mem
    for i in range(torch.cuda.device_count()					):
        lowerCamelCase__       : List[str]   					=						(torch.cuda.memory_allocated(_UpperCAmelCase					) - start_measures[str(_UpperCAmelCase					)]) / 2**20
        lowerCamelCase__       : Optional[Any]   					=						(torch.cuda.max_memory_allocated(_UpperCAmelCase					) - start_measures[str(_UpperCAmelCase					)]) / 2**20

    return measures


def 			SCREAMING_SNAKE_CASE (			_UpperCAmelCase		,      _UpperCAmelCase					)     ->	Dict:
    print(F"""{description}:"""					)
    print(F"""- Time: {measures["time"]:.2f}s"""					)
    for i in range(torch.cuda.device_count()					):
        print(F"""- GPU {i} allocated: {measures[str(_UpperCAmelCase					)]:.2f}MiB"""					)
        lowerCamelCase__       : List[str]   					=						measures[F"""{i}-peak"""]
        print(F"""- GPU {i} peak: {peak:.2f}MiB"""					)
    print(F"""- CPU RAM allocated: {measures["cpu"]:.2f}MiB"""					)
    print(F"""- CPU RAM peak: {measures["cpu-peak"]:.2f}MiB"""					)





