
'''simple docstring'''





import importlib
import importlib.metadata
import os
import warnings
from distutils.util import strtobool
from functools import lru_cache

import torch
from packaging import version
from packaging.version import parse

from .environment import parse_flag_from_env
from .versions import compare_versions, is_torch_version


try:
     import torch_xla.core.xla_model as xm  # noqa: F401

     A							     = True
except ImportError:
     A							     = False


# Cache this result has it's a C FFI call which can be pretty time-consuming
A							     = torch.distributed.is_available()
def 	snake_case_							(				_a  :				int      ):
      # Check we're not importing a "pkg_name" directory somewhere but the actual library by trying to grab the version
      snake_case_    							=				importlib.util.find_spec(_a      ) is not None
      if package_exists:
            try:
                  snake_case_    							=				importlib.metadata.metadata(_a      )
                  return True
            except importlib.metadata.PackageNotFoundError:
                  return False
def 	snake_case_							(				):
      return _torch_distributed_available
def 	snake_case_							(				):
      try:
            pass
      except ImportError:
            print(
                "Intel(R) oneCCL Bindings for PyTorch* is required to run DDP on Intel(R) GPUs, but it is not"
                " detected. If you see \"ValueError: Invalid backend: 'ccl'\" error, please install Intel(R) oneCCL"
                " Bindings for PyTorch*."      )
      return (
          importlib.util.find_spec("torch_ccl"      ) is not None
          or importlib.util.find_spec("oneccl_bindings_for_pytorch"      ) is not None
      )
def 	snake_case_							(				):
      return importlib.metadata.version("oneccl_bind_pt"      )
def 	snake_case_							(				):
      return _is_package_available("transformer_engine"      )
@lru_cache
def 	snake_case_							(				_a  :				str=True      ):
      # Due to bugs on the amp series GPUs, we disable torch-xla on them
      if torch.cuda.is_available():
            return False
      if _tpu_available and check_device:
            try:
                  # Will raise a RuntimeError if no XLA configuration is found
                  snake_case_    							=				xm.xla_device()
                  return True
            except RuntimeError:
                  return False
      return _tpu_available
def 	snake_case_							(				):
      return _is_package_available("deepspeed"      )
def 	snake_case_							(				_a  :				List[str]=False      ):
      if is_tpu_available():
            return not ignore_tpu
      if torch.cuda.is_available():
            return torch.cuda.is_bfaa_supported()
      if is_npu_available():
            return False
      return True
def 	snake_case_							(				):
      snake_case_    							=				_is_package_available("bitsandbytes"      )
      if package_exists:
            snake_case_    							=				version.parse(importlib.metadata.version("bitsandbytes"      )      )
            return compare_versions(_a			,       ">="			,       "0.39.0"      )
      return False
def 	snake_case_							(				):
      snake_case_    							=				_is_package_available("bitsandbytes"      )
      if package_exists:
            snake_case_    							=				version.parse(importlib.metadata.version("bitsandbytes"      )      )
            return compare_versions(_a			,       ">="			,       "0.37.2"      )
      return False
def 	snake_case_							(				):
      return _is_package_available("bitsandbytes"      )
def 	snake_case_							(				):
      if strtobool(os.environ.get("ACCELERATE_USE_MEGATRON_LM"			,       "False"      )      ) == 1:
            snake_case_    							=				importlib.util.find_spec("megatron"      ) is not None
            if package_exists:
                  try:
                        snake_case_    							=				parse(importlib.metadata.version("megatron-lm"      )      )
                        return compare_versions(_a			,       ">="			,       "2.2.0"      )
                  except Exception as e:
                        warnings.warn(F'''Parse Megatron version failed. Exception:{e}'''      )
                        return False
def 	snake_case_							(				):
      return _is_package_available("safetensors"      )
def 	snake_case_							(				):
      return _is_package_available("transformers"      )
def 	snake_case_							(				):
      return _is_package_available("datasets"      )
def 	snake_case_							(				):
      return _is_package_available("aim"      )
def 	snake_case_							(				):
      return _is_package_available("tensorboard"      ) or _is_package_available("tensorboardX"      )
def 	snake_case_							(				):
      return _is_package_available("wandb"      )
def 	snake_case_							(				):
      return _is_package_available("comet_ml"      )
def 	snake_case_							(				):
      return _is_package_available("boto3"      )
def 	snake_case_							(				):
      if _is_package_available("rich"      ):
            if "ACCELERATE_DISABLE_RICH" in os.environ:
                  warnings.warn(
                      "`ACCELERATE_DISABLE_RICH` is deprecated and will be removed in v0.22.0 and deactivated by default. Please use `ACCELERATE_ENABLE_RICH` if you wish to use `rich`."      )
                  return not parse_flag_from_env("ACCELERATE_DISABLE_RICH"			,       _a      )
            return parse_flag_from_env("ACCELERATE_ENABLE_RICH"			,       _a      )
      return False
def 	snake_case_							(				):
      return _is_package_available("sagemaker"      )
def 	snake_case_							(				):
      return _is_package_available("tqdm"      )
def 	snake_case_							(				):
      return _is_package_available("mlflow"      )
def 	snake_case_							(				):
      return is_torch_version(">="			,       "1.12"      ) and torch.backends.mps.is_available() and torch.backends.mps.is_built()
def 	snake_case_							(				):
      def get_major_and_minor_from_version(_a  :				Union[str, Any]      ):
            return str(version.parse(_a      ).major      ) + "." + str(version.parse(_a      ).minor      )

      snake_case_    							=				importlib.metadata.version("torch"      )
      if importlib.util.find_spec("intel_extension_for_pytorch"      ) is None:
            return False
      snake_case_    							=				"N/A"
      try:
            snake_case_    							=				importlib.metadata.version("intel_extension_for_pytorch"      )
      except importlib.metadata.PackageNotFoundError:
            return False
      snake_case_    							=				get_major_and_minor_from_version(_torch_version      )
      snake_case_    							=				get_major_and_minor_from_version(_ipex_version      )
      if torch_major_and_minor != ipex_major_and_minor:
            warnings.warn(
                F'''Intel Extension for PyTorch {ipex_major_and_minor} needs to work with PyTorch {ipex_major_and_minor}.*,'''
                F''' but PyTorch {_torch_version} is found. Please switch to the matching version and run again.'''      )
            return False
      return True
@lru_cache
def 	snake_case_							(				_a  :				Union[str, Any]=False      ):
      if importlib.util.find_spec("torch"      ) is None or importlib.util.find_spec("torch_npu"      ) is None:
            return False

      import torch
      import torch_npu  # noqa: F401

      if check_device:
            try:
                  # Will raise a RuntimeError if no NPU is found
                  snake_case_    							=				torch.npu.device_count()
                  return torch.npu.is_available()
            except RuntimeError:
                  return False
      return hasattr(_a			,       "npu"      ) and torch.npu.is_available()




@lru_cache
def 	snake_case_							(				_a  :				Optional[Any]=False      ):
      if not parse_flag_from_env("ACCELERATE_USE_XPU"			,       default=_a      ):
            return False
      if is_ipex_available():
            import torch

            if is_torch_version("<="			,       "1.12"      ):
                  return False
      else:
            return False

      import intel_extension_for_pytorch  # noqa: F401

      if check_device:
            try:
                  # Will raise a RuntimeError if no XPU  is found
                  snake_case_    							=				torch.xpu.device_count()
                  return torch.xpu.is_available()
            except RuntimeError:
                  return False
      return hasattr(_a			,       "xpu"      ) and torch.xpu.is_available()








"""simple docstring"""
import unittest

import numpy as np

from transformers.testing_utils import require_torch, require_vision
from transformers.utils import is_torch_available, is_vision_available

from ...test_image_processing_common import ImageProcessingSavingTestMixin


if is_torch_available():
		import torch

if is_vision_available():
		from PIL import Image

		from transformers import BlipImageProcessor






class    _UpperCAmelCase  (      unittest.TestCase					):


					def __init__(   self     :Optional[Any]	,			__UpperCamelCase     :Optional[int]	,			__UpperCamelCase     :Dict=7	,			__UpperCamelCase     :Optional[Any]=3	,			__UpperCamelCase     :Tuple=18	,			__UpperCamelCase     :List[Any]=30	,			__UpperCamelCase     :Optional[Any]=4_00	,			__UpperCamelCase     :Union[str, Any]=True	,			__UpperCamelCase     :Tuple=None	,			__UpperCamelCase     :List[str]=True	,			__UpperCamelCase     :Any=False	,			__UpperCamelCase     :str=[0.48_145_466, 0.4_578_275, 0.40_821_073]	,			__UpperCamelCase     :Optional[Any]=[0.26_862_954, 0.26_130_258, 0.27_577_711]	,			__UpperCamelCase     :Optional[Any]=True	,			):
												A						=    size if size is not None else {"height": 20, "width": 20}
												A						=    parent
												A						=    batch_size
												A						=    num_channels
												A						=    image_size
												A						=    min_resolution
												A						=    max_resolution
												A						=    do_resize
												A						=    size
												A						=    do_normalize
												A						=    image_mean
												A						=    image_std
												A						=    do_pad
												A						=    do_convert_rgb


					def       lowerCamelCase			(   self     :List[str]						):
												return {
												    "do_resize": self.do_resize,
												    "size": self.size,
												    "do_normalize": self.do_normalize,
												    "image_mean": self.image_mean,
												    "image_std": self.image_std,
												    "do_convert_rgb": self.do_convert_rgb,
												    "do_pad": self.do_pad,
												}


					def       lowerCamelCase			(   self     :Optional[Any]	,			__UpperCamelCase     :Optional[int]=False	,			__UpperCamelCase     :Dict=False	,			__UpperCamelCase     :Dict=False						):
												assert not (numpify and torchify), "You cannot specify both numpy and PyTorch tensors at the same time"

												if equal_resolution:
																			A						=    []
																			for i in range(self.batch_size						):
																										image_inputs.append(
																										    np.random.randint(
																										        2_55	,			size=(self.num_channels, self.max_resolution, self.max_resolution)	,			dtype=np.uinta						)						)
												else:
																			A						=    []
																			for i in range(self.batch_size						):
																										A,      A						=    np.random.choice(np.arange(self.min_resolution	,			self.max_resolution						)	,			2						)
																										image_inputs.append(np.random.randint(2_55	,			size=(self.num_channels, width, height)	,			dtype=np.uinta						)						)

												if not numpify and not torchify:
																			# PIL expects the channel dimension as last dimension
																			A						=    [Image.fromarray(np.moveaxis(__UpperCamelCase	,			0	,			-1						)						) for x in image_inputs]

												if torchify:
																			A						=    [torch.from_numpy(__UpperCamelCase						) for x in image_inputs]

												return image_inputs







@require_torch
@require_vision
class    _UpperCAmelCase  (      lowercase_	,				unittest.TestCase					):
					UpperCamelCase						=				BlipImageProcessor if is_vision_available() else None


					def       lowerCamelCase			(   self     :Optional[Any]						):
												A						=    BlipImageProcessingTester(self						)


					@property
					def       lowerCamelCase			(   self     :List[str]						):
												return self.image_processor_tester.prepare_image_processor_dict()


					def       lowerCamelCase			(   self     :Any						):
												A						=    self.image_processing_class(**self.image_processor_dict						)
												self.assertTrue(hasattr(__UpperCamelCase	,			"do_resize"						)						)
												self.assertTrue(hasattr(__UpperCamelCase	,			"size"						)						)
												self.assertTrue(hasattr(__UpperCamelCase	,			"do_normalize"						)						)
												self.assertTrue(hasattr(__UpperCamelCase	,			"image_mean"						)						)
												self.assertTrue(hasattr(__UpperCamelCase	,			"image_std"						)						)
												self.assertTrue(hasattr(__UpperCamelCase	,			"do_convert_rgb"						)						)


					def       lowerCamelCase			(   self     :List[Any]						):
												pass


					def       lowerCamelCase			(   self     :Optional[Any]						):
												# Initialize image_processor
												A						=    self.image_processing_class(**self.image_processor_dict						)
												# create random PIL images
												A						=    self.image_processor_tester.prepare_inputs(equal_resolution=__UpperCamelCase						)
												for image in image_inputs:
																			self.assertIsInstance(__UpperCamelCase	,			Image.Image						)

												# Test not batched input
												A						=    image_processor(image_inputs[0]	,			return_tensors="pt"						).pixel_values
												self.assertEqual(
												    encoded_images.shape	,			(
												        1,
												        self.image_processor_tester.num_channels,
												        self.image_processor_tester.size["height"],
												        self.image_processor_tester.size["width"],
												    )	,			)

												# Test batched
												A						=    image_processor(__UpperCamelCase	,			return_tensors="pt"						).pixel_values
												self.assertEqual(
												    encoded_images.shape	,			(
												        self.image_processor_tester.batch_size,
												        self.image_processor_tester.num_channels,
												        self.image_processor_tester.size["height"],
												        self.image_processor_tester.size["width"],
												    )	,			)


					def       lowerCamelCase			(   self     :Union[str, Any]						):
												# Initialize image_processor
												A						=    self.image_processing_class(**self.image_processor_dict						)
												# create random numpy tensors
												A						=    self.image_processor_tester.prepare_inputs(equal_resolution=__UpperCamelCase	,			numpify=__UpperCamelCase						)
												for image in image_inputs:
																			self.assertIsInstance(__UpperCamelCase	,			np.ndarray						)

												# Test not batched input
												A						=    image_processor(image_inputs[0]	,			return_tensors="pt"						).pixel_values
												self.assertEqual(
												    encoded_images.shape	,			(
												        1,
												        self.image_processor_tester.num_channels,
												        self.image_processor_tester.size["height"],
												        self.image_processor_tester.size["width"],
												    )	,			)

												# Test batched
												A						=    image_processor(__UpperCamelCase	,			return_tensors="pt"						).pixel_values
												self.assertEqual(
												    encoded_images.shape	,			(
												        self.image_processor_tester.batch_size,
												        self.image_processor_tester.num_channels,
												        self.image_processor_tester.size["height"],
												        self.image_processor_tester.size["width"],
												    )	,			)


					def       lowerCamelCase			(   self     :Tuple						):
												# Initialize image_processor
												A						=    self.image_processing_class(**self.image_processor_dict						)
												# create random PyTorch tensors
												A						=    self.image_processor_tester.prepare_inputs(equal_resolution=__UpperCamelCase	,			torchify=__UpperCamelCase						)
												for image in image_inputs:
																			self.assertIsInstance(__UpperCamelCase	,			torch.Tensor						)

												# Test not batched input
												A						=    image_processor(image_inputs[0]	,			return_tensors="pt"						).pixel_values
												self.assertEqual(
												    encoded_images.shape	,			(
												        1,
												        self.image_processor_tester.num_channels,
												        self.image_processor_tester.size["height"],
												        self.image_processor_tester.size["width"],
												    )	,			)

												# Test batched
												A						=    image_processor(__UpperCamelCase	,			return_tensors="pt"						).pixel_values
												self.assertEqual(
												    encoded_images.shape	,			(
												        self.image_processor_tester.batch_size,
												        self.image_processor_tester.num_channels,
												        self.image_processor_tester.size["height"],
												        self.image_processor_tester.size["width"],
												    )	,			)







@require_torch
@require_vision
class    _UpperCAmelCase  (      lowercase_	,				unittest.TestCase					):
					UpperCamelCase						=				BlipImageProcessor if is_vision_available() else None


					def       lowerCamelCase			(   self     :int						):
												A						=    BlipImageProcessingTester(self	,			num_channels=4						)
												A						=    3


					@property
					def       lowerCamelCase			(   self     :Optional[Any]						):
												return self.image_processor_tester.prepare_image_processor_dict()


					def       lowerCamelCase			(   self     :int						):
												A						=    self.image_processing_class(**self.image_processor_dict						)
												self.assertTrue(hasattr(__UpperCamelCase	,			"do_resize"						)						)
												self.assertTrue(hasattr(__UpperCamelCase	,			"size"						)						)
												self.assertTrue(hasattr(__UpperCamelCase	,			"do_normalize"						)						)
												self.assertTrue(hasattr(__UpperCamelCase	,			"image_mean"						)						)
												self.assertTrue(hasattr(__UpperCamelCase	,			"image_std"						)						)
												self.assertTrue(hasattr(__UpperCamelCase	,			"do_convert_rgb"						)						)


					def       lowerCamelCase			(   self     :Any						):
												pass


					def       lowerCamelCase			(   self     :Optional[Any]						):
												# Initialize image_processor
												A						=    self.image_processing_class(**self.image_processor_dict						)
												# create random PIL images
												A						=    self.image_processor_tester.prepare_inputs(equal_resolution=__UpperCamelCase						)
												for image in image_inputs:
																			self.assertIsInstance(__UpperCamelCase	,			Image.Image						)

												# Test not batched input
												A						=    image_processor(image_inputs[0]	,			return_tensors="pt"						).pixel_values
												self.assertEqual(
												    encoded_images.shape	,			(
												        1,
												        self.expected_encoded_image_num_channels,
												        self.image_processor_tester.size["height"],
												        self.image_processor_tester.size["width"],
												    )	,			)

												# Test batched
												A						=    image_processor(__UpperCamelCase	,			return_tensors="pt"						).pixel_values
												self.assertEqual(
												    encoded_images.shape	,			(
												        self.image_processor_tester.batch_size,
												        self.expected_encoded_image_num_channels,
												        self.image_processor_tester.size["height"],
												        self.image_processor_tester.size["width"],
												    )	,			)


