




import gc
import unittest

import numpy as np
import torch

from diffusers import RePaintPipeline, RePaintScheduler, UNetaDModel
from diffusers.utils.testing_utils import (
    enable_full_determinism,
    load_image,
    load_numpy,
    nightly,
    require_torch_gpu,
    skip_mps,
    torch_device,
)

from ..pipeline_params import IMAGE_INPAINTING_BATCH_PARAMS, IMAGE_INPAINTING_PARAMS
from ..test_pipelines_common import PipelineTesterMixin


enable_full_determinism()
class      a__       (   snake_case__					,							unittest.TestCase ):
						_a				:       Tuple   =       RePaintPipeline
						_a				:       Tuple   =       IMAGE_INPAINTING_PARAMS - {'width', 'height', 'guidance_scale'}
						_a				:       str   =       PipelineTesterMixin.required_optional_params - {
						    'latents',
						    'num_images_per_prompt',
						    'callback',
						    'callback_steps',
						}
						_a				:       Optional[Any]   =       IMAGE_INPAINTING_BATCH_PARAMS
						_a				:       Union[str, Any]   =       False






						def 							__SCREAMING_SNAKE_CASE(  self     ):




											"""simple docstring"""



											torch.manual_seed(0     )
											torch.manual_seed(0     )
											__lowerCAmelCase							  =			UNetaDModel(
											    block_out_channels=(3_2, 6_4)						,     layers_per_block=2						,     sample_size=3_2						,     in_channels=3						,     out_channels=3						,     down_block_types=("DownBlock2D", "AttnDownBlock2D")						,     up_block_types=("AttnUpBlock2D", "UpBlock2D")						,     )
											__lowerCAmelCase							  =			RePaintScheduler()
											__lowerCAmelCase							  =			{'''unet''': unet, '''scheduler''': scheduler}
											return components






						def 							__SCREAMING_SNAKE_CASE(  self						,     _A						,     _A=0     ):




											"""simple docstring"""



											if str(__snake_case     ).startswith("mps"     ):
																__lowerCAmelCase							  =			torch.manual_seed(__snake_case     )
											else:
																__lowerCAmelCase							  =			torch.Generator(device=__snake_case     ).manual_seed(__snake_case     )
											__lowerCAmelCase							  =			np.random.RandomState(__snake_case     ).standard_normal((1, 3, 3_2, 3_2)     )
											__lowerCAmelCase							  =			torch.from_numpy(__snake_case     ).to(device=__snake_case						,     dtype=torch.floataa     )
											__lowerCAmelCase							  =			(image > 0).to(device=__snake_case						,     dtype=torch.floataa     )
											__lowerCAmelCase							  =			{
											    '''image''': image,
											    '''mask_image''': mask,
											    '''generator''': generator,
											    '''num_inference_steps''': 5,
											    '''eta''': 0.0,
											    '''jump_length''': 2,
											    '''jump_n_sample''': 2,
											    '''output_type''': '''numpy''',
											}
											return inputs






						def 							__SCREAMING_SNAKE_CASE(  self     ):




											"""simple docstring"""



											__lowerCAmelCase							  =			'''cpu'''  # ensure determinism for the device-dependent torch.Generator
											__lowerCAmelCase							  =			self.get_dummy_components()
											__lowerCAmelCase							  =			RePaintPipeline(**__snake_case     )
											__lowerCAmelCase							  =			sd_pipe.to(__snake_case     )
											sd_pipe.set_progress_bar_config(disable=__snake_case     )

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

											assert image.shape == (1, 3_2, 3_2, 3)
											__lowerCAmelCase							  =			np.array([1.00_00, 0.54_26, 0.54_97, 0.22_00, 1.00_00, 1.00_00, 0.56_23, 1.00_00, 0.62_74]     )

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






						@skip_mps
						def 							__SCREAMING_SNAKE_CASE(  self     ):




											"""simple docstring"""



											return super().test_save_load_local()






						@unittest.skip("non-deterministic pipeline"     )
						def 							__SCREAMING_SNAKE_CASE(  self     ):




											"""simple docstring"""



											return super().test_inference_batch_single_identical()






						@skip_mps
						def 							__SCREAMING_SNAKE_CASE(  self     ):




											"""simple docstring"""



											return super().test_dict_tuple_outputs_equivalent()






						@skip_mps
						def 							__SCREAMING_SNAKE_CASE(  self     ):




											"""simple docstring"""



											return super().test_save_load_optional_components()






						@skip_mps
						def 							__SCREAMING_SNAKE_CASE(  self     ):




											"""simple docstring"""



											return super().test_attention_slicing_forward_pass()






@nightly
@require_torch_gpu
class      a__       (   unittest.TestCase ):






						def 							__SCREAMING_SNAKE_CASE(  self     ):




											"""simple docstring"""



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






						def 							__SCREAMING_SNAKE_CASE(  self     ):




											"""simple docstring"""



											__lowerCAmelCase							  =			load_image(
											    "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/"
											    "repaint/celeba_hq_256.png"     )
											__lowerCAmelCase							  =			load_image(
											    "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/repaint/mask_256.png"     )
											__lowerCAmelCase							  =			load_numpy(
											    "https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/"
											    "repaint/celeba_hq_256_result.npy"     )

											__lowerCAmelCase							  =			'''google/ddpm-ema-celebahq-256'''
											__lowerCAmelCase							  =			UNetaDModel.from_pretrained(__snake_case     )
											__lowerCAmelCase							  =			RePaintScheduler.from_pretrained(__snake_case     )

											__lowerCAmelCase							  =			RePaintPipeline(unet=__snake_case						,     scheduler=__snake_case     ).to(__snake_case     )
											repaint.set_progress_bar_config(disable=__snake_case     )
											repaint.enable_attention_slicing()

											__lowerCAmelCase							  =			torch.manual_seed(0     )
											__lowerCAmelCase							  =			repaint(
											    __snake_case						,     __snake_case						,     num_inference_steps=2_5_0						,     eta=0.0						,     jump_length=1_0						,     jump_n_sample=1_0						,     generator=__snake_case						,     output_type="np"						,     )
											__lowerCAmelCase							  =			output.images[0]

											assert image.shape == (2_5_6, 2_5_6, 3)
											assert np.abs(expected_image - image     ).mean() < 1E-2








import os
import random
from pathlib import Path
from typing import List

import numpy as np
import torch
from torch.cuda.amp import GradScaler

from .utils import (
    MODEL_NAME,
    OPTIMIZER_NAME,
    RNG_STATE_NAME,
    SCALER_NAME,
    SCHEDULER_NAME,
    get_pretty_name,
    is_tpu_available,
    is_xpu_available,
    save,
)


if is_tpu_available(check_device=False):
       import torch_xla.core.xla_model as xm

from .logging import get_logger
from .state import PartialState


__UpperCamelCase					:	int          =				get_logger(__name__)
def   snake_case_    (			__lowercase      ,							__lowercase      ,							__lowercase      ,							__lowercase      ,							__lowercase      ,							__lowercase = None      ,							):
   # Model states
   for i, state in enumerate(__lowercase							):
      UpperCAmelCase_				:   Dict							= F'''{MODEL_NAME}.bin''' if i == 0 else F'''{MODEL_NAME}_{i}.bin'''
      UpperCAmelCase_				:   List[Any]							= os.path.join(__lowercase      ,							__lowercase							)
      save(__lowercase      ,							__lowercase							)
      logger.info(F'''Model weights saved in {output_model_file}'''							)
   # Optimizer states
   for i, opt in enumerate(__lowercase							):
      UpperCAmelCase_				:   Tuple							= opt.state_dict()
      UpperCAmelCase_				:   Any							= F'''{OPTIMIZER_NAME}.bin''' if i == 0 else F'''{OPTIMIZER_NAME}_{i}.bin'''
      UpperCAmelCase_				:   List[Any]							= os.path.join(__lowercase      ,							__lowercase							)
      save(__lowercase      ,							__lowercase							)
      logger.info(F'''Optimizer state saved in {output_optimizer_file}'''							)
   # Scheduler states
   for i, scheduler in enumerate(__lowercase							):
      UpperCAmelCase_				:   Tuple							= scheduler.state_dict()
      UpperCAmelCase_				:   Tuple							= F'''{SCHEDULER_NAME}.bin''' if i == 0 else F'''{SCHEDULER_NAME}_{i}.bin'''
      UpperCAmelCase_				:   Tuple							= os.path.join(__lowercase      ,							__lowercase							)
      save(__lowercase      ,							__lowercase							)
      logger.info(F'''Scheduler state saved in {output_scheduler_file}'''							)
   # GradScaler state
   if scaler is not None:
      UpperCAmelCase_				:   Union[str, Any]							= scaler.state_dict()
      UpperCAmelCase_				:   Union[str, Any]							= os.path.join(__lowercase      ,							__lowercase							)
      torch.save(__lowercase      ,							__lowercase							)
      logger.info(F'''Gradient scaler state saved in {output_scaler_file}'''							)
   # Random number generator states
   UpperCAmelCase_				:   Union[str, Any]							= {}
   UpperCAmelCase_				:   Tuple							= F'''{RNG_STATE_NAME}_{process_index}.pkl'''
   UpperCAmelCase_				:   List[Any]							= random.getstate()
   UpperCAmelCase_				:   Any							= np.random.get_state()
   UpperCAmelCase_				:   List[Any]							= torch.get_rng_state()
   if is_xpu_available():
      UpperCAmelCase_				:   Optional[int]							= torch.xpu.get_rng_state_all()
   else:
      UpperCAmelCase_				:   Any							= torch.cuda.get_rng_state_all()
   if is_tpu_available():
      UpperCAmelCase_				:   List[str]							= xm.get_rng_state()
   UpperCAmelCase_				:   Any							= os.path.join(__lowercase      ,							__lowercase							)
   torch.save(__lowercase      ,							__lowercase							)
   logger.info(F'''Random states saved in {output_states_file}'''							)
   return output_dir
def   snake_case_    (			__lowercase      ,							__lowercase      ,							__lowercase      ,							__lowercase      ,							__lowercase      ,							__lowercase=None      ,							__lowercase=None      ,							**__lowercase      ,							):
   if map_location not in [None, "cpu", "on_device"]:
      raise TypeError(
          '''Unsupported optimizer map location passed, please choose one of `None`, `\'cpu\'`, or `\'on_device\'`'''							)
   if map_location is None:
      UpperCAmelCase_				:   Optional[Any]							= '''cpu'''
   elif map_location == "on_device":
      UpperCAmelCase_				:   Optional[int]							= PartialState().device
   # Model states
   for i, model in enumerate(__lowercase							):
      UpperCAmelCase_				:   List[Any]							= F'''{MODEL_NAME}.bin''' if i == 0 else F'''{MODEL_NAME}_{i}.bin'''
      UpperCAmelCase_				:   int							= os.path.join(__lowercase      ,							__lowercase							)
      models[i].load_state_dict(torch.load(__lowercase      ,							map_location=__lowercase							)      ,							**__lowercase							)
   logger.info('''All model weights loaded successfully'''							)

   # Optimizer states
   for i, opt in enumerate(__lowercase							):
      UpperCAmelCase_				:   List[str]							= F'''{OPTIMIZER_NAME}.bin''' if i == 0 else F'''{OPTIMIZER_NAME}_{i}.bin'''
      UpperCAmelCase_				:   List[Any]							= os.path.join(__lowercase      ,							__lowercase							)
      UpperCAmelCase_				:   Tuple							= torch.load(__lowercase      ,							map_location=__lowercase							)
      optimizers[i].load_state_dict(__lowercase							)
   logger.info('''All optimizer states loaded successfully'''							)

   # Scheduler states
   for i, scheduler in enumerate(__lowercase							):
      UpperCAmelCase_				:   Dict							= F'''{SCHEDULER_NAME}.bin''' if i == 0 else F'''{SCHEDULER_NAME}_{i}.bin'''
      UpperCAmelCase_				:   int							= os.path.join(__lowercase      ,							__lowercase							)
      scheduler.load_state_dict(torch.load(__lowercase							)							)
   logger.info('''All scheduler states loaded successfully'''							)

   # GradScaler state
   if scaler is not None:
      UpperCAmelCase_				:   List[str]							= os.path.join(__lowercase      ,							__lowercase							)
      scaler.load_state_dict(torch.load(__lowercase							)							)
      logger.info('''GradScaler state loaded successfully'''							)

   # Random states
   try:
      UpperCAmelCase_				:   Optional[Any]							= torch.load(os.path.join(__lowercase      ,							F'''{RNG_STATE_NAME}_{process_index}.pkl'''							)							)
      random.setstate(states['''random_state''']							)
      np.random.set_state(states['''numpy_random_seed''']							)
      torch.set_rng_state(states['''torch_manual_seed''']							)
      if is_xpu_available():
         torch.xpu.set_rng_state_all(states['''torch_xpu_manual_seed''']							)
      else:
         torch.cuda.set_rng_state_all(states['''torch_cuda_manual_seed''']							)
      if is_tpu_available():
         xm.set_rng_state(states['''xm_seed''']							)
      logger.info('''All random states loaded successfully'''							)
   except Exception:
      logger.info('''Could not load random states'''							)
def   snake_case_    (			__lowercase      ,							__lowercase      ,							__lowercase = 0							):
   UpperCAmelCase_				:   Optional[int]							= Path(__lowercase							) / F'''custom_checkpoint_{index}.pkl'''
   logger.info(F'''Saving the state of {get_pretty_name(__lowercase							)} to {save_location}'''							)
   torch.save(obj.state_dict()      ,							__lowercase							)


def   snake_case_    (			__lowercase      ,							__lowercase      ,							__lowercase = 0							):
   UpperCAmelCase_				:   Dict							= F'''{path}/custom_checkpoint_{index}.pkl'''
   logger.info(F'''Loading the state of {get_pretty_name(__lowercase							)} from {load_location}'''							)
   obj.load_state_dict(torch.load(__lowercase      ,							map_location='''cpu'''							)							)