





"""simple docstring"""





# Copyright 2021 The HuggingFace Team. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

from pathlib import Path

import torch

from ...utils import is_npu_available, is_xpu_available
from .config_args import ClusterConfig, default_json_config_file
from .config_utils import SubcommandHelpFormatter


__A											=		"""Create a default config file for Accelerate with only a few flags set."""
def  __A   (_SCREAMING_SNAKE_CASE="no"     ,   _SCREAMING_SNAKE_CASE = default_json_config_file     ,   _SCREAMING_SNAKE_CASE = False			)					->Any:





  """simple docstring"""


  lowerCAmelCase__			:Any											=							Path(_SCREAMING_SNAKE_CASE			)
  path.parent.mkdir(parents=_SCREAMING_SNAKE_CASE     ,   exist_ok=_SCREAMING_SNAKE_CASE			)
  if path.exists():
    print(
        F"Configuration already exists at {save_location}, will not override. Run `accelerate config` manually or pass a different `save_location`."			)
    return False
  lowerCAmelCase__			:Optional[Any]											=							mixed_precision.lower()
  if mixed_precision not in ["no", "fp16", "bf16", "fp8"]:
    raise ValueError(
        F"`mixed_precision` should be one of \'no\', \'fp16\', \'bf16\', or \'fp8\'. Received {mixed_precision}"			)
  lowerCAmelCase__			:Union[str, Any]											=							{
      '''compute_environment''': '''LOCAL_MACHINE''',
      '''mixed_precision''': mixed_precision,
  }
  if torch.cuda.is_available():
    lowerCAmelCase__			:List[str]											=							torch.cuda.device_count()
    lowerCAmelCase__			:Tuple											=							num_gpus
    lowerCAmelCase__			:Any											=							False
    if num_gpus > 1:
      lowerCAmelCase__			:Optional[Any]											=							'''MULTI_GPU'''
    else:
      lowerCAmelCase__			:List[Any]											=							'''NO'''
  elif is_xpu_available() and use_xpu:
    lowerCAmelCase__			:Optional[int]											=							torch.xpu.device_count()
    lowerCAmelCase__			:str											=							num_xpus
    lowerCAmelCase__			:int											=							False
    if num_xpus > 1:
      lowerCAmelCase__			:List[str]											=							'''MULTI_XPU'''
    else:
      lowerCAmelCase__			:Union[str, Any]											=							'''NO'''
  elif is_npu_available():
    lowerCAmelCase__			:Union[str, Any]											=							torch.npu.device_count()
    lowerCAmelCase__			:str											=							num_npus
    lowerCAmelCase__			:Any											=							False
    if num_npus > 1:
      lowerCAmelCase__			:str											=							'''MULTI_NPU'''
    else:
      lowerCAmelCase__			:Dict											=							'''NO'''
  else:
    lowerCAmelCase__			:List[Any]											=							0
    lowerCAmelCase__			:Any											=							True
    lowerCAmelCase__			:str											=							1
    lowerCAmelCase__			:Union[str, Any]											=							'''NO'''
  lowerCAmelCase__			:Optional[Any]											=							ClusterConfig(**_SCREAMING_SNAKE_CASE			)
  config.to_json_file(_SCREAMING_SNAKE_CASE			)
  return path
def  __A   (_SCREAMING_SNAKE_CASE     ,   _SCREAMING_SNAKE_CASE			)					->List[str]:





  """simple docstring"""


  lowerCAmelCase__			:int											=							parser.add_parser('default'     ,   parents=_SCREAMING_SNAKE_CASE     ,   help=_SCREAMING_SNAKE_CASE     ,   formatter_class=_SCREAMING_SNAKE_CASE			)
  parser.add_argument(
      '--config_file'     ,   default=_SCREAMING_SNAKE_CASE     ,   help=(
          'The path to use to store the config file. Will default to a file named default_config.yaml in the cache '
          'location, which is the content of the environment `HF_HOME` suffixed with \'accelerate\', or if you don\'t have '
          'such an environment variable, your cache directory (\'~/.cache\' or the content of `XDG_CACHE_HOME`) suffixed '
          'with \'huggingface\'.'
      )     ,   dest='save_location'     ,   )

  parser.add_argument(
      '--mixed_precision'     ,   choices=['no', 'fp16', 'bf16']     ,   type=_SCREAMING_SNAKE_CASE     ,   help='Whether or not to use mixed precision training. '
      'Choose between FP16 and BF16 (bfloat16) training. '
      'BF16 training is only supported on Nvidia Ampere GPUs and PyTorch 1.10 or later.'     ,   default='no'     ,   )
  parser.set_defaults(func=_SCREAMING_SNAKE_CASE			)
  return parser




def  __A   (_SCREAMING_SNAKE_CASE			)					->Dict:





  """simple docstring"""


  lowerCAmelCase__			:str											=							write_basic_config(args.mixed_precision     ,   args.save_location			)
  if config_file:
    print(F"accelerate configuration saved at {config_file}"			)




from typing import Dict, List, Optional, Union

import numpy as np

from ...image_processing_utils import BaseImageProcessor, BatchFeature, get_size_dict
from ...image_transforms import (
    center_crop,
    convert_to_rgb,
    get_resize_output_image_size,
    normalize,
    rescale,
    resize,
    to_channel_dimension_format,
)
from ...image_utils import (
    OPENAI_CLIP_MEAN,
    OPENAI_CLIP_STD,
    ChannelDimension,
    ImageInput,
    PILImageResampling,
    make_list_of_images,
    to_numpy_array,
    valid_images,
)
from ...utils import TensorType, is_vision_available, logging


UpperCamelCase								=	logging.get_logger(__name__)


if is_vision_available():
     import PIL






class   _lowerCamelCase (					UpperCamelCase			):




       """simple docstring"""
       snake_case      =    ["pixel_values"]



       def __init__(      self				,       _SCREAMING_SNAKE_CASE = True				,       _SCREAMING_SNAKE_CASE = None				,       _SCREAMING_SNAKE_CASE = PILImageResampling.BICUBIC				,       _SCREAMING_SNAKE_CASE = True				,       _SCREAMING_SNAKE_CASE = None				,       _SCREAMING_SNAKE_CASE = True				,       _SCREAMING_SNAKE_CASE = 1 / 255				,       _SCREAMING_SNAKE_CASE = True				,       _SCREAMING_SNAKE_CASE = None				,       _SCREAMING_SNAKE_CASE = None				,       _SCREAMING_SNAKE_CASE = True				,       **_SCREAMING_SNAKE_CASE				,       )->None:




         '''simple docstring'''





         super().__init__(**_SCREAMING_SNAKE_CASE		)
         A_      :				Tuple		     =					size if size is not None else {'''shortest_edge''': 224}
         A_      :				Any		     =					get_size_dict(_SCREAMING_SNAKE_CASE				,       default_to_square=_SCREAMING_SNAKE_CASE		)
         A_      :				Tuple		     =					crop_size if crop_size is not None else {'''height''': 224, '''width''': 224}
         A_      :				Dict		     =					get_size_dict(_SCREAMING_SNAKE_CASE				,       default_to_square=_SCREAMING_SNAKE_CASE				,       param_name='''crop_size'''		)

         A_      :				str		     =					do_resize
         A_      :				Tuple		     =					size
         A_      :				Optional[Any]		     =					resample
         A_      :				Tuple		     =					do_center_crop
         A_      :				List[Any]		     =					crop_size
         A_      :				Optional[int]		     =					do_rescale
         A_      :				Tuple		     =					rescale_factor
         A_      :				Any		     =					do_normalize
         A_      :				int		     =					image_mean if image_mean is not None else OPENAI_CLIP_MEAN
         A_      :				Any		     =					image_std if image_std is not None else OPENAI_CLIP_STD
         A_      :				Any		     =					do_convert_rgb



       def        _snake_case       (      self				,       _SCREAMING_SNAKE_CASE				,       _SCREAMING_SNAKE_CASE				,       _SCREAMING_SNAKE_CASE = PILImageResampling.BICUBIC				,       _SCREAMING_SNAKE_CASE = None				,       **_SCREAMING_SNAKE_CASE				,       )->np.ndarray:




         '''simple docstring'''





         A_      :				Dict		     =					get_size_dict(_SCREAMING_SNAKE_CASE				,       default_to_square=_SCREAMING_SNAKE_CASE		)
         if "shortest_edge" not in size:
           raise ValueError(F'''The `size` parameter must contain the key `shortest_edge`. Got {size.keys()}'''		)
         A_      :				Any		     =					get_resize_output_image_size(_SCREAMING_SNAKE_CASE				,       size=size['''shortest_edge''']				,       default_to_square=_SCREAMING_SNAKE_CASE		)
         return resize(_SCREAMING_SNAKE_CASE				,       size=_SCREAMING_SNAKE_CASE				,       resample=_SCREAMING_SNAKE_CASE				,       data_format=_SCREAMING_SNAKE_CASE				,       **_SCREAMING_SNAKE_CASE		)



       def        _snake_case       (      self				,       _SCREAMING_SNAKE_CASE				,       _SCREAMING_SNAKE_CASE				,       _SCREAMING_SNAKE_CASE = None				,       **_SCREAMING_SNAKE_CASE				,       )->np.ndarray:




         '''simple docstring'''





         A_      :				str		     =					get_size_dict(_SCREAMING_SNAKE_CASE		)
         if "height" not in size or "width" not in size:
           raise ValueError(F'''The `size` parameter must contain the keys (height, width). Got {size.keys()}'''		)
         return center_crop(_SCREAMING_SNAKE_CASE				,       size=(size['''height'''], size['''width'''])				,       data_format=_SCREAMING_SNAKE_CASE				,       **_SCREAMING_SNAKE_CASE		)



       def        _snake_case       (      self				,       _SCREAMING_SNAKE_CASE				,       _SCREAMING_SNAKE_CASE				,       _SCREAMING_SNAKE_CASE = None				,       **_SCREAMING_SNAKE_CASE				,       )->int:




         '''simple docstring'''





         return rescale(_SCREAMING_SNAKE_CASE				,       scale=_SCREAMING_SNAKE_CASE				,       data_format=_SCREAMING_SNAKE_CASE				,       **_SCREAMING_SNAKE_CASE		)



       def        _snake_case       (      self				,       _SCREAMING_SNAKE_CASE				,       _SCREAMING_SNAKE_CASE				,       _SCREAMING_SNAKE_CASE				,       _SCREAMING_SNAKE_CASE = None				,       **_SCREAMING_SNAKE_CASE				,       )->np.ndarray:




         '''simple docstring'''





         return normalize(_SCREAMING_SNAKE_CASE				,       mean=_SCREAMING_SNAKE_CASE				,       std=_SCREAMING_SNAKE_CASE				,       data_format=_SCREAMING_SNAKE_CASE				,       **_SCREAMING_SNAKE_CASE		)







       def        _snake_case       (      self				,       _SCREAMING_SNAKE_CASE				,       _SCREAMING_SNAKE_CASE = None				,       _SCREAMING_SNAKE_CASE = None				,       _SCREAMING_SNAKE_CASE = None				,       _SCREAMING_SNAKE_CASE = None				,       _SCREAMING_SNAKE_CASE = None				,       _SCREAMING_SNAKE_CASE = None				,       _SCREAMING_SNAKE_CASE = None				,       _SCREAMING_SNAKE_CASE = None				,       _SCREAMING_SNAKE_CASE = None				,       _SCREAMING_SNAKE_CASE = None				,       _SCREAMING_SNAKE_CASE = None				,       _SCREAMING_SNAKE_CASE = None				,       _SCREAMING_SNAKE_CASE = ChannelDimension.FIRST				,       **_SCREAMING_SNAKE_CASE				,       )->PIL.Image.Image:




         '''simple docstring'''





         A_      :				Optional[int]		     =					do_resize if do_resize is not None else self.do_resize
         A_      :				int		     =					size if size is not None else self.size
         A_      :				Optional[int]		     =					get_size_dict(_SCREAMING_SNAKE_CASE				,       param_name='''size'''				,       default_to_square=_SCREAMING_SNAKE_CASE		)
         A_      :				List[Any]		     =					resample if resample is not None else self.resample
         A_      :				Any		     =					do_center_crop if do_center_crop is not None else self.do_center_crop
         A_      :				List[str]		     =					crop_size if crop_size is not None else self.crop_size
         A_      :				int		     =					get_size_dict(_SCREAMING_SNAKE_CASE				,       param_name='''crop_size'''				,       default_to_square=_SCREAMING_SNAKE_CASE		)
         A_      :				Optional[int]		     =					do_rescale if do_rescale is not None else self.do_rescale
         A_      :				int		     =					rescale_factor if rescale_factor is not None else self.rescale_factor
         A_      :				int		     =					do_normalize if do_normalize is not None else self.do_normalize
         A_      :				Tuple		     =					image_mean if image_mean is not None else self.image_mean
         A_      :				Tuple		     =					image_std if image_std is not None else self.image_std
         A_      :				List[str]		     =					do_convert_rgb if do_convert_rgb is not None else self.do_convert_rgb

         A_      :				Optional[int]		     =					make_list_of_images(_SCREAMING_SNAKE_CASE		)

         if not valid_images(_SCREAMING_SNAKE_CASE		):
           raise ValueError(
               '''Invalid image type. Must be of type PIL.Image.Image, numpy.ndarray, '''
               '''torch.Tensor, tf.Tensor or jax.ndarray.'''		)

         if do_resize and size is None:
           raise ValueError('''Size must be specified if do_resize is True.'''		)

         if do_center_crop and crop_size is None:
           raise ValueError('''Crop size must be specified if do_center_crop is True.'''		)

         if do_rescale and rescale_factor is None:
           raise ValueError('''Rescale factor must be specified if do_rescale is True.'''		)

         if do_normalize and (image_mean is None or image_std is None):
           raise ValueError('''Image mean and std must be specified if do_normalize is True.'''		)

         # PIL RGBA images are converted to RGB
         if do_convert_rgb:
           A_      :				List[str]		     =					[convert_to_rgb(_SCREAMING_SNAKE_CASE		) for image in images]

         # All transformations expect numpy arrays.
         A_      :				int		     =					[to_numpy_array(_SCREAMING_SNAKE_CASE		) for image in images]

         if do_resize:
           A_      :				Tuple		     =					[self.resize(image=_SCREAMING_SNAKE_CASE				,       size=_SCREAMING_SNAKE_CASE				,       resample=_SCREAMING_SNAKE_CASE		) for image in images]

         if do_center_crop:
           A_      :				Union[str, Any]		     =					[self.center_crop(image=_SCREAMING_SNAKE_CASE				,       size=_SCREAMING_SNAKE_CASE		) for image in images]

         if do_rescale:
           A_      :				Tuple		     =					[self.rescale(image=_SCREAMING_SNAKE_CASE				,       scale=_SCREAMING_SNAKE_CASE		) for image in images]

         if do_normalize:
           A_      :				List[Any]		     =					[self.normalize(image=_SCREAMING_SNAKE_CASE				,       mean=_SCREAMING_SNAKE_CASE				,       std=_SCREAMING_SNAKE_CASE		) for image in images]

         A_      :				str		     =					[to_channel_dimension_format(_SCREAMING_SNAKE_CASE				,       _SCREAMING_SNAKE_CASE		) for image in images]

         A_      :				Optional[Any]		     =					{'''pixel_values''': images}
         return BatchFeature(data=_SCREAMING_SNAKE_CASE				,       tensor_type=_SCREAMING_SNAKE_CASE		)




