

import json
import logging
import os
import socket

import git
import numpy as np
import torch


logging.basicConfig(
    format='''%(asctime)s - %(levelname)s - %(name)s - PID: %(process)d -  %(message)s''',
    datefmt='''%m/%d/%Y %H:%M:%S''',
    level=logging.INFO,
)
lowerCamelCase           =					logging.getLogger(__name__)




def 			UpperCAmelCase__				(			_A      : str							):





  '''simple docstring'''







  a__  =git.Repo(search_parent_directories=_A							)
  a__  ={
      '''repo_id''': str(_A							),
      '''repo_sha''': str(repo.head.object.hexsha							),
      '''repo_branch''': str(repo.active_branch							),
  }

  with open(os.path.join(_A	,      '''git_log.json'''							)	,      '''w'''							) as f:
    json.dump(_A	,      _A	,      indent=4							)




def 			UpperCAmelCase__				(			_A      : str							):





  '''simple docstring'''







  if params.n_gpu <= 0:
    a__  =0
    a__  =-1
    a__  =True
    a__  =False
    return

  assert torch.cuda.is_available()

  logger.info('''Initializing GPUs'''							)
  if params.n_gpu > 1:
    assert params.local_rank != -1

    a__  =int(os.environ['''WORLD_SIZE''']							)
    a__  =int(os.environ['''N_GPU_NODE''']							)
    a__  =int(os.environ['''RANK''']							)

    # number of nodes / node ID
    a__  =params.world_size // params.n_gpu_per_node
    a__  =params.global_rank // params.n_gpu_per_node
    a__  =True

    assert params.n_nodes == int(os.environ['''N_NODES''']							)
    assert params.node_id == int(os.environ['''NODE_RANK''']							)

  # local job (single GPU)
  else:
    assert params.local_rank == -1

    a__  =1
    a__  =0
    a__  =0
    a__  =0
    a__  =1
    a__  =1
    a__  =False

  # sanity checks
  assert params.n_nodes >= 1
  assert 0 <= params.node_id < params.n_nodes
  assert 0 <= params.local_rank <= params.global_rank < params.world_size
  assert params.world_size == params.n_nodes * params.n_gpu_per_node

  # define whether this is the master process / if we are in multi-node distributed mode
  a__  =params.node_id == 0 and params.local_rank == 0
  a__  =params.n_nodes > 1

  # summary
  a__  =F"""--- Global rank: {params.global_rank} - """
  logger.info(PREFIX + '''Number of nodes: %i''' % params.n_nodes							)
  logger.info(PREFIX + '''Node ID        : %i''' % params.node_id							)
  logger.info(PREFIX + '''Local rank     : %i''' % params.local_rank							)
  logger.info(PREFIX + '''World size     : %i''' % params.world_size							)
  logger.info(PREFIX + '''GPUs per node  : %i''' % params.n_gpu_per_node							)
  logger.info(PREFIX + '''Master         : %s''' % str(params.is_master							)							)
  logger.info(PREFIX + '''Multi-node     : %s''' % str(params.multi_node							)							)
  logger.info(PREFIX + '''Multi-GPU      : %s''' % str(params.multi_gpu							)							)
  logger.info(PREFIX + '''Hostname       : %s''' % socket.gethostname()							)

  # set GPU device
  torch.cuda.set_device(params.local_rank							)

  # initialize multi-GPU
  if params.multi_gpu:
    logger.info('''Initializing PyTorch distributed'''							)
    torch.distributed.init_process_group(
        init_method='''env://'''	,      backend='''nccl'''	,      )




def 			UpperCAmelCase__				(			_A      : Any							):





  '''simple docstring'''







  np.random.seed(args.seed							)
  torch.manual_seed(args.seed							)
  if args.n_gpu > 0:
    torch.cuda.manual_seed_all(args.seed							)



import random
import unittest

import torch

from diffusers import IFInpaintingSuperResolutionPipeline
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_INPAINTING_BATCH_PARAMS,
    TEXT_GUIDED_IMAGE_INPAINTING_PARAMS,
)
from ..test_pipelines_common import PipelineTesterMixin
from . import IFPipelineTesterMixin







@skip_mps
class 						__magic_name__			(							lowerCamelCase__		,       lowerCamelCase__		,       unittest.TestCase							):

   '''simple docstring'''




   lowerCamelCase__ :	List[Any]    =   IFInpaintingSuperResolutionPipeline
   lowerCamelCase__ :	str    =   TEXT_GUIDED_IMAGE_INPAINTING_PARAMS - {'width', 'height'}
   lowerCamelCase__ :	List[Any]    =   TEXT_GUIDED_IMAGE_INPAINTING_BATCH_PARAMS.union({'original_image'}							)
   lowerCamelCase__ :	str    =   PipelineTesterMixin.required_optional_params - {'latents'}



   def  _UpperCAmelCase (				self       )							->    Optional[int]:






     """simple docstring"""






     return self._get_superresolution_dummy_components()



   def  _UpperCAmelCase (				self,    lowercase_,    lowercase_=0       )							->    Tuple:






     """simple docstring"""






     if str(lowercase_       ).startswith('''mps'''       ):
       a__  =torch.manual_seed(lowercase_       )
     else:
       a__  =torch.Generator(device=lowercase_       ).manual_seed(lowercase_       )

     a__  =floats_tensor((1, 3, 16, 16),    rng=random.Random(lowercase_       )       ).to(lowercase_       )
     a__  =floats_tensor((1, 3, 32, 32),    rng=random.Random(lowercase_       )       ).to(lowercase_       )
     a__  =floats_tensor((1, 3, 32, 32),    rng=random.Random(lowercase_       )       ).to(lowercase_       )

     a__  ={
         '''prompt''': '''A painting of a squirrel eating a burger''',
         '''image''': image,
         '''original_image''': original_image,
         '''mask_image''': mask_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  _UpperCAmelCase (				self       )							->    List[str]:






     """simple docstring"""






     self._test_xformers_attention_forwardGenerator_pass(expected_max_diff=1E-3       )



   def  _UpperCAmelCase (				self       )							->    int:






     """simple docstring"""






     self._test_save_load_optional_components()



   @unittest.skipIf(torch_device != '''cuda''',    reason='''float16 requires CUDA'''       )
   def  _UpperCAmelCase (				self       )							->    List[Any]:






     """simple docstring"""






     super().test_save_load_floataa(expected_max_diff=1E-1       )



   def  _UpperCAmelCase (				self       )							->    int:






     """simple docstring"""






     self._test_attention_slicing_forward_pass(expected_max_diff=1E-2       )



   def  _UpperCAmelCase (				self       )							->    Optional[Any]:






     """simple docstring"""






     self._test_save_load_local()



   def  _UpperCAmelCase (				self       )							->    str:






     """simple docstring"""






     self._test_inference_batch_single_identical(
         expected_max_diff=1E-2,    )
