






import gc
import threading
import time

import psutil
import torch
class     a    :



      """simple docstring"""


      def __init__(      self					:			int					)							->      Tuple:
        __UpperCAmelCase							:	Tuple           =  psutil.Process()
        __UpperCAmelCase							:	str           =  False
      def 	UpperCAmelCase      (      self					:			List[str]					)							->      Optional[Any]:
        __UpperCAmelCase							:	Optional[int]           =  -1

        while True:
          __UpperCAmelCase							:	Optional[int]           =  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 	UpperCAmelCase      (      self					:			Any					)							->      Any:
        __UpperCAmelCase							:	Optional[int]           =  True
        __UpperCAmelCase							:	Optional[Any]           =  threading.Thread(target=self.peak_monitor					)
        __UpperCAmelCase							:	Tuple           =  True
        self.thread.start()




      def 	UpperCAmelCase      (      self					:			Optional[int]					)							->      int:
        __UpperCAmelCase							:	List[Any]           =  False
        self.thread.join()
        return self.cpu_memory_peak


a		:       str			     = PeakCPUMemory()






def 				lowerCamelCase__      (    ):
  # Time
  __UpperCAmelCase							:	List[Any]           =  {"""time""": time.time()}

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

  # CPU mem
  __UpperCAmelCase							:	Any           =  psutil.Process().memory_info().rss
  cpu_peak_tracker.start()

  # GPU mem
  for i in range(torch.cuda.device_count()		):
    __UpperCAmelCase							:	Any           =  torch.cuda.memory_allocated(__lowerCamelCase		)
  torch.cuda.reset_peak_memory_stats()

  return measures






def 				lowerCamelCase__      (    __lowerCamelCase			:     Tuple		):
  # Time
  __UpperCAmelCase							:	Any           =  {"""time""": time.time() - start_measures["""time"""]}

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

  # CPU mem
  __UpperCAmelCase							:	Optional[int]           =  (psutil.Process().memory_info().rss - start_measures["""cpu"""]) / 2**20
  __UpperCAmelCase							:	str           =  (cpu_peak_tracker.stop() - start_measures["""cpu"""]) / 2**20

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

  return measures







def 				lowerCamelCase__      (    __lowerCamelCase			:     Union[str, Any]  ,		__lowerCamelCase			:     Any		):
  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(__lowerCamelCase		)]:.2f}MiB"""		)
    __UpperCAmelCase							:	Optional[int]           =  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"""		)










import random
import unittest

import numpy as np

from diffusers import (
    DPMSolverMultistepScheduler,
    EulerAncestralDiscreteScheduler,
    EulerDiscreteScheduler,
    LMSDiscreteScheduler,
    OnnxStableDiffusionImgaImgPipeline,
    PNDMScheduler,
)
from diffusers.utils import floats_tensor
from diffusers.utils.testing_utils import (
    is_onnx_available,
    load_image,
    nightly,
    require_onnxruntime,
    require_torch_gpu,
)

from ..test_pipelines_onnx_common import OnnxPipelineTesterMixin


if is_onnx_available():
       import onnxruntime as ort
class     a    ( lowercase__							,       unittest.TestCase   ):



      """simple docstring"""


      a  : int							       =	'hf-internal-testing/tiny-random-OnnxStableDiffusionPipeline'
      def 	UpperCAmelCase      (      self					:			List[str]		,      __lowercase					:			Optional[Any]=0					)							->      Any:
        __UpperCAmelCase							:	Any           =  floats_tensor((1, 3, 128, 128)		,      rng=random.Random(__lowercase					)					)
        __UpperCAmelCase							:	int           =  np.random.RandomState(__lowercase					)
        __UpperCAmelCase							:	Optional[Any]           =  {
            """prompt""": """A painting of a squirrel eating a burger""",
            """image""": image,
            """generator""": generator,
            """num_inference_steps""": 3,
            """strength""": 0.75,
            """guidance_scale""": 7.5,
            """output_type""": """numpy""",
        }
        return inputs
      def 	UpperCAmelCase      (      self					:			Optional[int]					)							->      Optional[Any]:
        __UpperCAmelCase							:	List[Any]           =  OnnxStableDiffusionImgaImgPipeline.from_pretrained(self.hub_checkpoint		,      provider="""CPUExecutionProvider"""					)
        pipe.set_progress_bar_config(disable=__lowercase					)

        __UpperCAmelCase							:	int           =  self.get_dummy_inputs()
        __UpperCAmelCase							:	Optional[Any]           =  pipe(**__lowercase					).images
        __UpperCAmelCase							:	Union[str, Any]           =  image[0, -3:, -3:, -1].flatten()

        assert image.shape == (1, 128, 128, 3)
        __UpperCAmelCase							:	List[str]           =  np.array([0.69_643, 0.58_484, 0.50_314, 0.58_760, 0.55_368, 0.59_643, 0.51_529, 0.41_217, 0.49_087]					)
        assert np.abs(image_slice - expected_slice					).max() < 1e-1
      def 	UpperCAmelCase      (      self					:			Optional[Any]					)							->      List[str]:
        __UpperCAmelCase							:	Dict           =  OnnxStableDiffusionImgaImgPipeline.from_pretrained(self.hub_checkpoint		,      provider="""CPUExecutionProvider"""					)
        __UpperCAmelCase							:	int           =  PNDMScheduler.from_config(pipe.scheduler.config		,      skip_prk_steps=__lowercase					)
        pipe.set_progress_bar_config(disable=__lowercase					)

        __UpperCAmelCase							:	Any           =  self.get_dummy_inputs()
        __UpperCAmelCase							:	Tuple           =  pipe(**__lowercase					).images
        __UpperCAmelCase							:	Dict           =  image[0, -3:, -3:, -1]

        assert image.shape == (1, 128, 128, 3)
        __UpperCAmelCase							:	str           =  np.array([0.61_737, 0.54_642, 0.53_183, 0.54_465, 0.52_742, 0.60_525, 0.49_969, 0.40_655, 0.48_154]					)

        assert np.abs(image_slice.flatten() - expected_slice					).max() < 1e-1
      def 	UpperCAmelCase      (      self					:			str					)							->      Tuple:
        __UpperCAmelCase							:	Tuple           =  OnnxStableDiffusionImgaImgPipeline.from_pretrained(self.hub_checkpoint		,      provider="""CPUExecutionProvider"""					)
        __UpperCAmelCase							:	int           =  LMSDiscreteScheduler.from_config(pipe.scheduler.config					)
        pipe.set_progress_bar_config(disable=__lowercase					)

        # warmup pass to apply optimizations
        __UpperCAmelCase							:	Optional[int]           =  pipe(**self.get_dummy_inputs()					)

        __UpperCAmelCase							:	Tuple           =  self.get_dummy_inputs()
        __UpperCAmelCase							:	Any           =  pipe(**__lowercase					).images
        __UpperCAmelCase							:	List[Any]           =  image[0, -3:, -3:, -1]

        assert image.shape == (1, 128, 128, 3)
        __UpperCAmelCase							:	Optional[int]           =  np.array([0.52_761, 0.59_977, 0.49_033, 0.49_619, 0.54_282, 0.50_311, 0.47_600, 0.40_918, 0.45_203]					)

        assert np.abs(image_slice.flatten() - expected_slice					).max() < 1e-1
      def 	UpperCAmelCase      (      self					:			Optional[Any]					)							->      str:
        __UpperCAmelCase							:	int           =  OnnxStableDiffusionImgaImgPipeline.from_pretrained(self.hub_checkpoint		,      provider="""CPUExecutionProvider"""					)
        __UpperCAmelCase							:	Union[str, Any]           =  EulerDiscreteScheduler.from_config(pipe.scheduler.config					)
        pipe.set_progress_bar_config(disable=__lowercase					)

        __UpperCAmelCase							:	List[str]           =  self.get_dummy_inputs()
        __UpperCAmelCase							:	int           =  pipe(**__lowercase					).images
        __UpperCAmelCase							:	Dict           =  image[0, -3:, -3:, -1]

        assert image.shape == (1, 128, 128, 3)
        __UpperCAmelCase							:	Tuple           =  np.array([0.52_911, 0.60_004, 0.49_229, 0.49_805, 0.54_502, 0.50_680, 0.47_777, 0.41_028, 0.45_304]					)

        assert np.abs(image_slice.flatten() - expected_slice					).max() < 1e-1
      def 	UpperCAmelCase      (      self					:			int					)							->      Any:
        __UpperCAmelCase							:	Any           =  OnnxStableDiffusionImgaImgPipeline.from_pretrained(self.hub_checkpoint		,      provider="""CPUExecutionProvider"""					)
        __UpperCAmelCase							:	List[Any]           =  EulerAncestralDiscreteScheduler.from_config(pipe.scheduler.config					)
        pipe.set_progress_bar_config(disable=__lowercase					)

        __UpperCAmelCase							:	Union[str, Any]           =  self.get_dummy_inputs()
        __UpperCAmelCase							:	int           =  pipe(**__lowercase					).images
        __UpperCAmelCase							:	List[Any]           =  image[0, -3:, -3:, -1]

        assert image.shape == (1, 128, 128, 3)
        __UpperCAmelCase							:	List[str]           =  np.array([0.52_911, 0.60_004, 0.49_229, 0.49_805, 0.54_502, 0.50_680, 0.47_777, 0.41_028, 0.45_304]					)

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




      def 	UpperCAmelCase      (      self					:			Tuple					)							->      str:
        __UpperCAmelCase							:	Dict           =  OnnxStableDiffusionImgaImgPipeline.from_pretrained(self.hub_checkpoint		,      provider="""CPUExecutionProvider"""					)
        __UpperCAmelCase							:	str           =  DPMSolverMultistepScheduler.from_config(pipe.scheduler.config					)
        pipe.set_progress_bar_config(disable=__lowercase					)

        __UpperCAmelCase							:	Optional[Any]           =  self.get_dummy_inputs()
        __UpperCAmelCase							:	int           =  pipe(**__lowercase					).images
        __UpperCAmelCase							:	List[Any]           =  image[0, -3:, -3:, -1]

        assert image.shape == (1, 128, 128, 3)
        __UpperCAmelCase							:	Union[str, Any]           =  np.array([0.65_331, 0.58_277, 0.48_204, 0.56_059, 0.53_665, 0.56_235, 0.50_969, 0.40_009, 0.46_552]					)

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


@nightly
@require_onnxruntime
@require_torch_gpu
class     a    ( unittest.TestCase   ):



      """simple docstring"""


      @property
      def 	UpperCAmelCase      (      self					:			Dict					)							->      List[Any]:
        return (
            "CUDAExecutionProvider",
            {
                "gpu_mem_limit": "15000000000",  # 15GB
                "arena_extend_strategy": "kSameAsRequested",
            },
        )
      @property
      def 	UpperCAmelCase      (      self					:			Tuple					)							->      Tuple:
        __UpperCAmelCase							:	Optional[int]           =  ort.SessionOptions()
        __UpperCAmelCase							:	List[Any]           =  False
        return options
      def 	UpperCAmelCase      (      self					:			List[str]					)							->      Tuple:
        __UpperCAmelCase							:	List[Any]           =  load_image(
            """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"""
            """/img2img/sketch-mountains-input.jpg"""					)
        __UpperCAmelCase							:	Dict           =  init_image.resize((768, 512)					)
        # using the PNDM scheduler by default
        __UpperCAmelCase							:	Any           =  OnnxStableDiffusionImgaImgPipeline.from_pretrained(
            """CompVis/stable-diffusion-v1-4"""		,      revision="""onnx"""		,      safety_checker=__lowercase		,      feature_extractor=__lowercase		,      provider=self.gpu_provider		,      sess_options=self.gpu_options		,      )
        pipe.set_progress_bar_config(disable=__lowercase					)

        __UpperCAmelCase							:	Dict           =  """A fantasy landscape, trending on artstation"""

        __UpperCAmelCase							:	str           =  np.random.RandomState(0					)
        __UpperCAmelCase							:	Optional[Any]           =  pipe(
            prompt=__lowercase		,      image=__lowercase		,      strength=0.75		,      guidance_scale=7.5		,      num_inference_steps=10		,      generator=__lowercase		,      output_type="""np"""		,      )
        __UpperCAmelCase							:	str           =  output.images
        __UpperCAmelCase							:	int           =  images[0, 255:258, 383:386, -1]

        assert images.shape == (1, 512, 768, 3)
        __UpperCAmelCase							:	Union[str, Any]           =  np.array([0.4_909, 0.5_059, 0.5_372, 0.4_623, 0.4_876, 0.5_049, 0.4_820, 0.4_956, 0.5_019]					)
        # TODO: lower the tolerance after finding the cause of onnxruntime reproducibility issues

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




      def 	UpperCAmelCase      (      self					:			Optional[Any]					)							->      str:
        __UpperCAmelCase							:	Tuple           =  load_image(
            """https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main"""
            """/img2img/sketch-mountains-input.jpg"""					)
        __UpperCAmelCase							:	int           =  init_image.resize((768, 512)					)
        __UpperCAmelCase							:	Tuple           =  LMSDiscreteScheduler.from_pretrained(
            """runwayml/stable-diffusion-v1-5"""		,      subfolder="""scheduler"""		,      revision="""onnx"""					)
        __UpperCAmelCase							:	Tuple           =  OnnxStableDiffusionImgaImgPipeline.from_pretrained(
            """runwayml/stable-diffusion-v1-5"""		,      revision="""onnx"""		,      scheduler=__lowercase		,      safety_checker=__lowercase		,      feature_extractor=__lowercase		,      provider=self.gpu_provider		,      sess_options=self.gpu_options		,      )
        pipe.set_progress_bar_config(disable=__lowercase					)

        __UpperCAmelCase							:	Dict           =  """A fantasy landscape, trending on artstation"""

        __UpperCAmelCase							:	int           =  np.random.RandomState(0					)
        __UpperCAmelCase							:	Optional[int]           =  pipe(
            prompt=__lowercase		,      image=__lowercase		,      strength=0.75		,      guidance_scale=7.5		,      num_inference_steps=20		,      generator=__lowercase		,      output_type="""np"""		,      )
        __UpperCAmelCase							:	Union[str, Any]           =  output.images
        __UpperCAmelCase							:	Union[str, Any]           =  images[0, 255:258, 383:386, -1]

        assert images.shape == (1, 512, 768, 3)
        __UpperCAmelCase							:	str           =  np.array([0.8_043, 0.926, 0.9_581, 0.8_119, 0.8_954, 0.913, 0.7_209, 0.7_463, 0.7_431]					)
        # TODO: lower the tolerance after finding the cause of onnxruntime reproducibility issues

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


