


# 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


UpperCamelCase								=	"""Create a default config file for Accelerate with only a few flags set."""


def   _SCREAMING_SNAKE_CASE	( SCREAMING_SNAKE_CASE="no" ,			SCREAMING_SNAKE_CASE = default_json_config_file ,			SCREAMING_SNAKE_CASE = False       ):
  A_      :				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
  A_      :				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}'''       )
  A_      :				Union[str, Any]		     =					{
      '''compute_environment''': '''LOCAL_MACHINE''',
      '''mixed_precision''': mixed_precision,
  }
  if torch.cuda.is_available():
    A_      :				List[str]		     =					torch.cuda.device_count()
    A_      :				Tuple		     =					num_gpus
    A_      :				Any		     =					False
    if num_gpus > 1:
      A_      :				Optional[Any]		     =					'''MULTI_GPU'''
    else:
      A_      :				List[Any]		     =					'''NO'''
  elif is_xpu_available() and use_xpu:
    A_      :				Optional[int]		     =					torch.xpu.device_count()
    A_      :				str		     =					num_xpus
    A_      :				int		     =					False
    if num_xpus > 1:
      A_      :				List[str]		     =					'''MULTI_XPU'''
    else:
      A_      :				Union[str, Any]		     =					'''NO'''
  elif is_npu_available():
    A_      :				Union[str, Any]		     =					torch.npu.device_count()
    A_      :				str		     =					num_npus
    A_      :				Any		     =					False
    if num_npus > 1:
      A_      :				str		     =					'''MULTI_NPU'''
    else:
      A_      :				Dict		     =					'''NO'''
  else:
    A_      :				List[Any]		     =					0
    A_      :				Any		     =					True
    A_      :				str		     =					1
    A_      :				Union[str, Any]		     =					'''NO'''
  A_      :				Optional[Any]		     =					ClusterConfig(**SCREAMING_SNAKE_CASE       )
  config.to_json_file(SCREAMING_SNAKE_CASE       )
  return path


def   _SCREAMING_SNAKE_CASE	( SCREAMING_SNAKE_CASE ,			SCREAMING_SNAKE_CASE       ):
  A_      :				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   _SCREAMING_SNAKE_CASE	( SCREAMING_SNAKE_CASE       ):
  A_      :				str		     =					write_basic_config(args.mixed_precision ,			args.save_location       )
  if config_file:
    print(f'''accelerate configuration saved at {config_file}'''       )








import os
import sys
import warnings
from dataclasses import dataclass, field
from io import BytesIO
from typing import TYPE_CHECKING, Any, ClassVar, Dict, List, Optional, Union

import numpy as np
import pyarrow as pa

from .. import config
from ..download.streaming_download_manager import xopen
from ..table import array_cast
from ..utils.file_utils import is_local_path
from ..utils.py_utils import first_non_null_value, no_op_if_value_is_null, string_to_dict


if TYPE_CHECKING:
     import PIL.Image

     from .features import FeatureType


UpperCamelCase								=	None
UpperCamelCase								=	"""<""" if sys.byteorder == """little""" else """>"""
# Origin: https://github.com/python-pillow/Pillow/blob/698951e19e19972aeed56df686868f1329981c12/src/PIL/Image.py#L3126 minus "|i1" which values are not preserved correctly when saving and loading an image
UpperCamelCase								=	[
    np.dtype("""|b1"""),
    np.dtype("""|u1"""),
    np.dtype("""<u2"""),
    np.dtype(""">u2"""),
    np.dtype("""<i2"""),
    np.dtype(""">i2"""),
    np.dtype("""<u4"""),
    np.dtype(""">u4"""),
    np.dtype("""<i4"""),
    np.dtype(""">i4"""),
    np.dtype("""<f4"""),
    np.dtype(""">f4"""),
    np.dtype("""<f8"""),
    np.dtype(""">f8"""),
]






@dataclass
class   _lowerCamelCase :




       """simple docstring"""
       snake_case      =    True
       snake_case      =    None
       # Automatically constructed
       snake_case      =    "PIL.Image.Image"
       snake_case      =    pa.struct({"bytes": pa.binary(), "path": pa.string()}			)
       snake_case      =    field(default="Image"					,							init=UpperCamelCase					,							repr=UpperCamelCase			)



       def __call__(      self		)->int:




         '''simple docstring'''





         return self.pa_type



       def        _snake_case       (      self				,       _SCREAMING_SNAKE_CASE		)->dict:




         '''simple docstring'''





         if config.PIL_AVAILABLE:
           import PIL.Image
         else:
           raise ImportError('''To support encoding images, please install \'Pillow\'.'''		)

         if isinstance(_SCREAMING_SNAKE_CASE				,       _SCREAMING_SNAKE_CASE		):
           A_      :				Optional[int]		     =					np.array(_SCREAMING_SNAKE_CASE		)

         if isinstance(_SCREAMING_SNAKE_CASE				,       _SCREAMING_SNAKE_CASE		):
           return {"path": value, "bytes": None}
         elif isinstance(_SCREAMING_SNAKE_CASE				,       _SCREAMING_SNAKE_CASE		):
           return {"path": None, "bytes": value}
         elif isinstance(_SCREAMING_SNAKE_CASE				,       np.ndarray		):
           # convert the image array to PNG/TIFF bytes
           return encode_np_array(_SCREAMING_SNAKE_CASE		)
         elif isinstance(_SCREAMING_SNAKE_CASE				,       PIL.Image.Image		):
           # convert the PIL image to bytes (default format is PNG/TIFF)
           return encode_pil_image(_SCREAMING_SNAKE_CASE		)
         elif value.get('''path'''		) is not None and os.path.isfile(value['''path''']		):
           # we set "bytes": None to not duplicate the data if they're already available locally
           return {"bytes": None, "path": value.get('''path'''		)}
         elif value.get('''bytes'''		) is not None or value.get('''path'''		) is not None:
           # store the image bytes, and path is used to infer the image format using the file extension
           return {"bytes": value.get('''bytes'''		), "path": value.get('''path'''		)}
         else:
           raise ValueError(
               F'''An image sample should have one of \'path\' or \'bytes\' but they are missing or None in {value}.'''		)



       def        _snake_case       (      self				,       _SCREAMING_SNAKE_CASE				,       _SCREAMING_SNAKE_CASE=None		)->"PIL.Image.Image":




         '''simple docstring'''





         if not self.decode:
           raise RuntimeError('''Decoding is disabled for this feature. Please use Image(decode=True) instead.'''		)

         if config.PIL_AVAILABLE:
           import PIL.Image
         else:
           raise ImportError('''To support decoding images, please install \'Pillow\'.'''		)

         if token_per_repo_id is None:
           A_      :				List[str]		     =					{}

         A_     ,       A_      :				str		     =					value['''path'''], value['''bytes''']
         if bytes_ is None:
           if path is None:
             raise ValueError(F'''An image should have one of \'path\' or \'bytes\' but both are None in {value}.'''		)
           else:
             if is_local_path(_SCREAMING_SNAKE_CASE		):
               A_      :				List[str]		     =					PIL.Image.open(_SCREAMING_SNAKE_CASE		)
             else:
               A_      :				List[str]		     =					path.split('''::'''		)[-1]
               try:
                 A_      :				int		     =					string_to_dict(_SCREAMING_SNAKE_CASE				,       config.HUB_DATASETS_URL		)['''repo_id''']
                 A_      :				Optional[int]		     =					token_per_repo_id.get(_SCREAMING_SNAKE_CASE		)
               except ValueError:
                 A_      :				Any		     =					None
               with xopen(_SCREAMING_SNAKE_CASE				,       '''rb'''				,       use_auth_token=_SCREAMING_SNAKE_CASE		) as f:
                 A_      :				Optional[Any]		     =					BytesIO(f.read()		)
               A_      :				Dict		     =					PIL.Image.open(bytes_		)
         else:
           A_      :				Optional[int]		     =					PIL.Image.open(BytesIO(bytes_		)		)
         image.load()  # to avoid "Too many open files" errors
         return image



       def        _snake_case       (      self		)->Union["FeatureType", Dict[str, "FeatureType"]]:




         '''simple docstring'''





         from .features import Value

         return (
             self
             if self.decode
             else {
                 "bytes": Value('''binary'''		),
                 "path": Value('''string'''		),
             }
         )



       def        _snake_case       (      self				,       _SCREAMING_SNAKE_CASE		)->pa.StructArray:




         '''simple docstring'''





         if pa.types.is_string(storage.type		):
           A_      :				Dict		     =					pa.array([None] * len(_SCREAMING_SNAKE_CASE		)				,       type=pa.binary()		)
           A_      :				List[Any]		     =					pa.StructArray.from_arrays([bytes_array, storage]				,       ['''bytes''', '''path''']				,       mask=storage.is_null()		)
         elif pa.types.is_binary(storage.type		):
           A_      :				Dict		     =					pa.array([None] * len(_SCREAMING_SNAKE_CASE		)				,       type=pa.string()		)
           A_      :				List[str]		     =					pa.StructArray.from_arrays([storage, path_array]				,       ['''bytes''', '''path''']				,       mask=storage.is_null()		)
         elif pa.types.is_struct(storage.type		):
           if storage.type.get_field_index('''bytes'''		) >= 0:
             A_      :				Tuple		     =					storage.field('''bytes'''		)
           else:
             A_      :				Optional[int]		     =					pa.array([None] * len(_SCREAMING_SNAKE_CASE		)				,       type=pa.binary()		)
           if storage.type.get_field_index('''path'''		) >= 0:
             A_      :				Optional[Any]		     =					storage.field('''path'''		)
           else:
             A_      :				Optional[int]		     =					pa.array([None] * len(_SCREAMING_SNAKE_CASE		)				,       type=pa.string()		)
           A_      :				Union[str, Any]		     =					pa.StructArray.from_arrays([bytes_array, path_array]				,       ['''bytes''', '''path''']				,       mask=storage.is_null()		)
         elif pa.types.is_list(storage.type		):
           A_      :				Optional[Any]		     =					pa.array(
               [encode_np_array(np.array(_SCREAMING_SNAKE_CASE		)		)['''bytes'''] if arr is not None else None for arr in storage.to_pylist()]				,       type=pa.binary()				,       )
           A_      :				str		     =					pa.array([None] * len(_SCREAMING_SNAKE_CASE		)				,       type=pa.string()		)
           A_      :				List[Any]		     =					pa.StructArray.from_arrays(
               [bytes_array, path_array]				,       ['''bytes''', '''path''']				,       mask=bytes_array.is_null()		)
         return array_cast(_SCREAMING_SNAKE_CASE				,       self.pa_type		)







       def        _snake_case       (      self				,       _SCREAMING_SNAKE_CASE		)->pa.StructArray:




         '''simple docstring'''





         @no_op_if_value_is_null
         def path_to_bytes(_SCREAMING_SNAKE_CASE		):
           with xopen(_SCREAMING_SNAKE_CASE				,       '''rb'''		) as f:
             A_      :				Any		     =					f.read()
           return bytes_

         A_      :				Dict		     =					pa.array(
             [
                 (path_to_bytes(x['''path''']		) if x['''bytes'''] is None else x['''bytes''']) if x is not None else None
                 for x in storage.to_pylist()
             ]				,       type=pa.binary()				,       )
         A_      :				List[Any]		     =					pa.array(
             [os.path.basename(_SCREAMING_SNAKE_CASE		) if path is not None else None for path in storage.field('''path'''		).to_pylist()]				,       type=pa.string()				,       )
         A_      :				str		     =					pa.StructArray.from_arrays([bytes_array, path_array]				,       ['''bytes''', '''path''']				,       mask=bytes_array.is_null()		)
         return array_cast(_SCREAMING_SNAKE_CASE				,       self.pa_type		)




def   _SCREAMING_SNAKE_CASE	( ):
  if config.PIL_AVAILABLE:
    import PIL.Image
  else:
    raise ImportError('''To support encoding images, please install \'Pillow\'.'''       )

  global _IMAGE_COMPRESSION_FORMATS
  if _IMAGE_COMPRESSION_FORMATS is None:
    PIL.Image.init()
    A_      :				Dict		     =					list(set(PIL.Image.OPEN.keys()       ) & set(PIL.Image.SAVE.keys()       )       )
  return _IMAGE_COMPRESSION_FORMATS


def   _SCREAMING_SNAKE_CASE	( SCREAMING_SNAKE_CASE       ):
  A_      :				Dict		     =					BytesIO()
  if image.format in list_image_compression_formats():
    A_      :				Tuple		     =					image.format
  else:
    A_      :				List[str]		     =					'''PNG''' if image.mode in ['''1''', '''L''', '''LA''', '''RGB''', '''RGBA'''] else '''TIFF'''
  image.save(SCREAMING_SNAKE_CASE ,			format=SCREAMING_SNAKE_CASE       )
  return buffer.getvalue()


def   _SCREAMING_SNAKE_CASE	( SCREAMING_SNAKE_CASE       ):
  if hasattr(SCREAMING_SNAKE_CASE ,			'''filename'''       ) and image.filename != "":
    return {"path": image.filename, "bytes": None}
  else:
    return {"path": None, "bytes": image_to_bytes(SCREAMING_SNAKE_CASE       )}


def   _SCREAMING_SNAKE_CASE	( SCREAMING_SNAKE_CASE       ):
  if config.PIL_AVAILABLE:
    import PIL.Image
  else:
    raise ImportError('''To support encoding images, please install \'Pillow\'.'''       )

  A_      :				Union[str, Any]		     =					array.dtype
  A_      :				Dict		     =					dtype.byteorder if dtype.byteorder != '''=''' else _NATIVE_BYTEORDER
  A_      :				Any		     =					dtype.kind
  A_      :				Any		     =					dtype.itemsize

  A_      :				Dict		     =					None

  # Multi-channel array case (only np.dtype("|u1") is allowed)
  if array.shape[2:]:
    A_      :				List[Any]		     =					np.dtype('''|u1'''       )
    if dtype_kind not in ["u", "i"]:
      raise TypeError(
          f'''Unsupported array dtype {dtype} for image encoding. Only {dest_dtype} is supported for multi-channel arrays.'''       )
    if dtype is not dest_dtype:
      warnings.warn(f'''Downcasting array dtype {dtype} to {dest_dtype} to be compatible with \'Pillow\''''       )
    # Exact match
  elif dtype in _VALID_IMAGE_ARRAY_DTPYES:
    A_      :				int		     =					dtype
  else:  # Downcast the type within the kind (np.can_cast(from_type, to_type, casting="same_kind") doesn't behave as expected, so do it manually)
    while dtype_itemsize >= 1:
      A_      :				Any		     =					dtype_byteorder + dtype_kind + str(SCREAMING_SNAKE_CASE       )
      A_      :				Optional[int]		     =					np.dtype(SCREAMING_SNAKE_CASE       )
      if dest_dtype in _VALID_IMAGE_ARRAY_DTPYES:
        warnings.warn(f'''Downcasting array dtype {dtype} to {dest_dtype} to be compatible with \'Pillow\''''       )
        break
      else:
        dtype_itemsize //= 2

  if dest_dtype is None:
    raise TypeError(
        f'''Cannot convert dtype {dtype} to a valid image dtype. Valid image dtypes: {_VALID_IMAGE_ARRAY_DTPYES}'''       )
  A_      :				Tuple		     =					PIL.Image.fromarray(array.astype(SCREAMING_SNAKE_CASE       )       )
  return {"path": None, "bytes": image_to_bytes(SCREAMING_SNAKE_CASE       )}







def   _SCREAMING_SNAKE_CASE	( SCREAMING_SNAKE_CASE       ):
  if config.PIL_AVAILABLE:
    import PIL.Image
  else:
    raise ImportError('''To support encoding images, please install \'Pillow\'.'''       )

  if objs:
    A_     ,       A_      :				Union[str, Any]		     =					first_non_null_value(SCREAMING_SNAKE_CASE       )
    if isinstance(SCREAMING_SNAKE_CASE ,			SCREAMING_SNAKE_CASE       ):
      return [{"path": obj, "bytes": None} if obj is not None else None for obj in objs]
    if isinstance(SCREAMING_SNAKE_CASE ,			np.ndarray       ):
      A_      :				Tuple		     =					no_op_if_value_is_null(SCREAMING_SNAKE_CASE       )
      return [obj_to_image_dict_func(SCREAMING_SNAKE_CASE       ) for obj in objs]
    elif isinstance(SCREAMING_SNAKE_CASE ,			PIL.Image.Image       ):
      A_      :				List[str]		     =					no_op_if_value_is_null(SCREAMING_SNAKE_CASE       )
      return [obj_to_image_dict_func(SCREAMING_SNAKE_CASE       ) for obj in objs]
    else:
      return objs
  else:
    return objs




