


import importlib.util
import json
import os
import warnings
from dataclasses import dataclass, field

import torch

from ..training_args import TrainingArguments
from ..utils import cached_property, is_sagemaker_dp_enabled, logging


a__    :							Optional[int]            =					logging.get_logger(__name__)




def        UpperCAmelCase__	():




					'''simple docstring'''


					__SCREAMING_SNAKE_CASE						       =    os.getenv("SM_HP_MP_PARAMETERS"  ,						"{}"       )
					try:
										# Parse it and check the field "partitions" is included, it is required for model parallel.
										__SCREAMING_SNAKE_CASE						       =    json.loads(SCREAMING_SNAKE_CASE__       )
										if "partitions" not in smp_options:
															return False
					except json.JSONDecodeError:
										return False

					# Get the sagemaker specific framework parameters from mpi_options variable.
					__SCREAMING_SNAKE_CASE						       =    os.getenv("SM_FRAMEWORK_PARAMS"  ,						"{}"       )
					try:
										# Parse it and check the field "sagemaker_distributed_dataparallel_enabled".
										__SCREAMING_SNAKE_CASE						       =    json.loads(SCREAMING_SNAKE_CASE__       )
										if not mpi_options.get("sagemaker_mpi_enabled"  ,						SCREAMING_SNAKE_CASE__       ):
															return False
					except json.JSONDecodeError:
										return False
					# Lastly, check if the `smdistributed` module is present.
					return importlib.util.find_spec("smdistributed"       ) is not None


if is_sagemaker_model_parallel_available():
		import smdistributed.modelparallel.torch as smp

		smp.init()




@dataclass
class 			UpperCamelCase_     (			lowerCamelCase_):

	"""simple docstring"""





	snake_case__    :  str    =      field(
	    default=""      ,  metadata={"help": "Used by the SageMaker launcher to send mp-specific args. Ignored in SageMakerTrainer"}      ,  )




	def   UpperCAmelCase_				(			self      :    Tuple       )	->     Optional[Any]:
						super().__post_init__()
						warnings.warn(
						    "`SageMakerTrainingArguments` is deprecated and will be removed in v5 of Transformers. You can use "
						    "`TrainingArguments` instead."      ,      _UpperCAmelCase      ,      )




	@cached_property
	def   UpperCAmelCase_				(			self      :    Union[str, Any]       )	->     Tuple:
						logger.info("PyTorch: setting up devices"       )
						if torch.distributed.is_available() and torch.distributed.is_initialized() and self.local_rank == -1:
											logger.warning(
											    "torch.distributed process group is initialized, but local_rank == -1. "
											    "In order to use Torch DDP, launch your script with `python -m torch.distributed.launch"       )
						if self.no_cuda:
											__SCREAMING_SNAKE_CASE						       =    torch.device("cpu"       )
											__SCREAMING_SNAKE_CASE						       =    0
						elif is_sagemaker_model_parallel_available():
											__SCREAMING_SNAKE_CASE						       =    smp.local_rank()
											__SCREAMING_SNAKE_CASE						       =    torch.device("cuda"      ,      _UpperCAmelCase       )
											__SCREAMING_SNAKE_CASE						       =    1
						elif is_sagemaker_dp_enabled():
											import smdistributed.dataparallel.torch.torch_smddp  # noqa: F401

											torch.distributed.init_process_group(backend="smddp"      ,      timeout=self.ddp_timeout_delta       )
											__SCREAMING_SNAKE_CASE						       =    int(os.getenv("SMDATAPARALLEL_LOCAL_RANK"       )       )
											__SCREAMING_SNAKE_CASE						       =    torch.device("cuda"      ,      self.local_rank       )
											__SCREAMING_SNAKE_CASE						       =    1
						elif self.local_rank == -1:
											# if n_gpu is > 1 we'll use nn.DataParallel.
											# If you only want to use a specific subset of GPUs use `CUDA_VISIBLE_DEVICES=0`
											# Explicitly set CUDA to the first (index 0) CUDA device, otherwise `set_device` will
											# trigger an error that a device index is missing. Index 0 takes into account the
											# GPUs available in the environment, so `CUDA_VISIBLE_DEVICES=1,2` with `cuda:0`
											# will use the first GPU in that env, i.e. GPU#1
											__SCREAMING_SNAKE_CASE						       =    torch.device("cuda:0" if torch.cuda.is_available() else "cpu"       )
											# Sometimes the line in the postinit has not been run before we end up here, so just checking we're not at
											# the default value.
											__SCREAMING_SNAKE_CASE						       =    torch.cuda.device_count()
						else:
											# Here, we'll use torch.distributed.
											# Initializes the distributed backend which will take care of synchronizing nodes/GPUs
											if not torch.distributed.is_initialized():
																torch.distributed.init_process_group(backend="nccl"      ,      timeout=self.ddp_timeout_delta       )
											__SCREAMING_SNAKE_CASE						       =    torch.device("cuda"      ,      self.local_rank       )
											__SCREAMING_SNAKE_CASE						       =    1

						if device.type == "cuda":
											torch.cuda.set_device(_UpperCAmelCase       )

						return device




	@property
	def   UpperCAmelCase_				(			self      :    List[Any]       )	->     int:






						if is_sagemaker_model_parallel_available():
											return smp.dp_size()

						return super().world_size




	@property
	def   UpperCAmelCase_				(			self      :    Dict       )	->     Optional[Any]:
						return not is_sagemaker_model_parallel_available()




	@property
	def   UpperCAmelCase_				(			self      :    Optional[int]       )	->     List[str]:
						return False




"""simple docstring"""







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 			UpperCamelCase_     (			UpperCamelCase      ,  UpperCamelCase      ,  unittest.TestCase):

	"""simple docstring"""





	snake_case__    :  str    =      IFInpaintingSuperResolutionPipeline
	snake_case__    :  Optional[Any]    =      TEXT_GUIDED_IMAGE_INPAINTING_PARAMS - {"width", "height"}
	snake_case__    :  Union[str, Any]    =      TEXT_GUIDED_IMAGE_INPAINTING_BATCH_PARAMS.union({"original_image"})
	snake_case__    :  Dict    =      PipelineTesterMixin.required_optional_params - {"latents"}




	def   UpperCAmelCase_				(			self      :    List[Any]       )	->     Optional[int]:
						return self._get_superresolution_dummy_components()




	def   UpperCAmelCase_				(			self      :    List[str]      ,      UpperCAmelCase__      :    List[Any]      ,      UpperCAmelCase__      :    List[str]=0       )	->     List[str]:
						if str(UpperCAmelCase__       ).startswith("mps"       ):
											__SCREAMING_SNAKE_CASE						       =    torch.manual_seed(UpperCAmelCase__       )
						else:
											__SCREAMING_SNAKE_CASE						       =    torch.Generator(device=UpperCAmelCase__       ).manual_seed(UpperCAmelCase__       )

						__SCREAMING_SNAKE_CASE						       =    floats_tensor((1, 3, 1_6, 1_6)      ,      rng=random.Random(UpperCAmelCase__       )       ).to(UpperCAmelCase__       )
						__SCREAMING_SNAKE_CASE						       =    floats_tensor((1, 3, 3_2, 3_2)      ,      rng=random.Random(UpperCAmelCase__       )       ).to(UpperCAmelCase__       )
						__SCREAMING_SNAKE_CASE						       =    floats_tensor((1, 3, 3_2, 3_2)      ,      rng=random.Random(UpperCAmelCase__       )       ).to(UpperCAmelCase__       )

						__SCREAMING_SNAKE_CASE						       =    {
						    "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      :    str       )	->     Dict:
						self._test_xformers_attention_forwardGenerator_pass(expected_max_diff=1E-3       )




	def   UpperCAmelCase_				(			self      :    str       )	->     Tuple:
						self._test_save_load_optional_components()




	@unittest.skipIf(torch_device != "cuda"      ,      reason="float16 requires CUDA"       )
	def   UpperCAmelCase_				(			self      :    Tuple       )	->     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   UpperCAmelCase_				(			self      :    Optional[Any]       )	->     List[Any]:
						self._test_attention_slicing_forward_pass(expected_max_diff=1E-2       )




	def   UpperCAmelCase_				(			self      :    Any       )	->     Union[str, Any]:
						self._test_save_load_local()




	def   UpperCAmelCase_				(			self      :    Union[str, Any]       )	->     Optional[Any]:
						self._test_inference_batch_single_identical(
						    expected_max_diff=1E-2      ,      )
