'''simple docstring'''






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


_UpperCAmelCase    :     Tuple         =       logging.get_logger(__name__)





def  _SCREAMING_SNAKE_CASE		(      ):
     # Get the sagemaker specific mp parameters from smp_options variable.
     _A    		=  os.getenv('SM_HP_MP_PARAMETERS'   ,		'{}'				)
     try:
          # Parse it and check the field "partitions" is included, it is required for model parallel.
          _A    		=  json.loads(__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.
     _A    		=  os.getenv('SM_FRAMEWORK_PARAMS'   ,		'{}'				)
     try:
          # Parse it and check the field "sagemaker_distributed_dataparallel_enabled".
          _A    		=  json.loads(__snake_case				)
          if not mpi_options.get('sagemaker_mpi_enabled'   ,		__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 lowercase_       (		_UpperCamelCase			):







   """simple docstring"""

   __lowerCAmelCase         =   field(
       default=""    ,    metadata={"help": "Used by the SageMaker launcher to send mp-specific args. Ignored in SageMakerTrainer"}    ,    )
   def 	__UpperCAmelCase   (      self      :			Any     )  ->			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      :			str     )  ->			"torch.device":
        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:
             _A    		=  torch.device('cpu'     )
             _A    		=  0
        elif is_sagemaker_model_parallel_available():
             _A    		=  smp.local_rank()
             _A    		=  torch.device('cuda', UpperCamelCase__     )
             _A    		=  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     )
             _A    		=  int(os.getenv('SMDATAPARALLEL_LOCAL_RANK'     )     )
             _A    		=  torch.device('cuda', self.local_rank     )
             _A    		=  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
             _A    		=  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.
             _A    		=  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     )
             _A    		=  torch.device('cuda', self.local_rank     )
             _A    		=  1

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

        return device
   @property
   def 	__UpperCAmelCase   (      self      :			Union[str, Any]     )  ->			Optional[Any]:
        if is_sagemaker_model_parallel_available():
             return smp.dp_size()

        return super().world_size
   @property
   def 	__UpperCAmelCase   (      self      :			Any     )  ->			int:
        return not is_sagemaker_model_parallel_available()

   @property
   def 	__UpperCAmelCase   (      self      :			Dict     )  ->			List[Any]:
        return False




'''simple docstring'''






import unittest

import numpy as np
import torch

from diffusers import VersatileDiffusionImageVariationPipeline
from diffusers.utils.testing_utils import load_image, require_torch_gpu, slow, torch_device


_UpperCAmelCase    :     List[Any]         =       False







class lowercase_       (		unittest.TestCase			):







   """simple docstring"""

   pass







@slow
@require_torch_gpu
class lowercase_       (		unittest.TestCase			):







   """simple docstring"""

   def 	__UpperCAmelCase   (      self      :			List[str]     )  ->			Dict:
        _A    		=  VersatileDiffusionImageVariationPipeline.from_pretrained('shi-labs/versatile-diffusion'     )
        pipe.to(UpperCamelCase__     )
        pipe.set_progress_bar_config(disable=UpperCamelCase__     )

        _A    		=  load_image(
            'https://huggingface.co/datasets/hf-internal-testing/diffusers-images/resolve/main/versatile_diffusion/benz.jpg'     )
        _A    		=  torch.manual_seed(0     )
        _A    		=  pipe(
            image=UpperCamelCase__, generator=UpperCamelCase__, guidance_scale=7.5, num_inference_steps=50, output_type='numpy', ).images

        _A    		=  image[0, 2_53:2_56, 2_53:2_56, -1]

        assert image.shape == (1, 5_12, 5_12, 3)
        _A    		=  np.array([0.0_441, 0.0_469, 0.0_507, 0.0_575, 0.0_632, 0.0_650, 0.0_865, 0.0_909, 0.0_945]     )

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



