


# 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


SCREAMING_SNAKE_CASE			: str				   =					"Create a default config file for Accelerate with only a few flags set."



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



def 							UpperCamelCase_( lowerCamelCase_				,					lowerCamelCase_					)       ->		List[Any]:
    _lowercase			:				Union[str, Any] 						=						parser.add_parser('default'				,					parents=lowerCamelCase_				,					help=lowerCamelCase_				,					formatter_class=lowerCamelCase_					)
    parser.add_argument(
        '--config_file'				,					default=lowerCamelCase_				,					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=lowerCamelCase_				,					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=lowerCamelCase_					)
    return parser



def 							UpperCamelCase_( lowerCamelCase_					)       ->		int:
    _lowercase			:				List[str] 						=						write_basic_config(args.mixed_precision				,					args.save_location					)
    if config_file:
        print(F'''accelerate configuration saved at {config_file}'''					)








from collections.abc import Iterator, MutableMapping
from dataclasses import dataclass
from typing import Generic, TypeVar

__magic_name__  :  Any					=       TypeVar("""KEY""")
__magic_name__  :  Dict					=       TypeVar("""VAL""")





@dataclass(frozen=_a    ,					slots=_a			)
class     SCREAMING_SNAKE_CASE__		(Generic[KEY, VAL]			):
     lowercase_ :      KEY
     lowercase_ :      VAL





class     SCREAMING_SNAKE_CASE__		(_Item			):
     def __init__(						self	:		Tuple						):



       """simple docstring"""
       super().__init__(__lowerCamelCase       ,						__lowerCamelCase						)



     def __bool__(						self	:		Tuple						):



       """simple docstring"""
       return False



__magic_name__  :  str					=       _DeletedItem()





class     SCREAMING_SNAKE_CASE__		(MutableMapping[KEY, VAL]			):
     def __init__(						self	:		Any       ,						__lowerCamelCase	:		int = 8       ,						__lowerCamelCase	:		float = 0.75						):



       """simple docstring"""
       lowerCAmelCase__										=      initial_block_size
       lowerCAmelCase__										=      [None] * initial_block_size
       assert 0.0 < capacity_factor < 1.0
       lowerCAmelCase__										=      capacity_factor
       lowerCAmelCase__										=      0
     def        A__      (						self	:		Union[str, Any]       ,						__lowerCamelCase	:		KEY						):



       """simple docstring"""
       return hash(__lowerCamelCase						) % len(self._buckets						)
     def        A__      (						self	:		Tuple       ,						__lowerCamelCase	:		int						):



       """simple docstring"""
       return (ind + 1) % len(self._buckets						)
     def        A__      (						self	:		Dict       ,						__lowerCamelCase	:		int       ,						__lowerCamelCase	:		KEY       ,						__lowerCamelCase	:		VAL						):



       """simple docstring"""
       lowerCAmelCase__										=      self._buckets[ind]
       if not stored:
         lowerCAmelCase__										=      _Item(__lowerCamelCase       ,						__lowerCamelCase						)
         self._len += 1
         return True
       elif stored.key == key:
         lowerCAmelCase__										=      _Item(__lowerCamelCase       ,						__lowerCamelCase						)
         return True
       else:
         return False
     def        A__      (						self	:		Any						):



       """simple docstring"""
       lowerCAmelCase__										=      len(self._buckets						) * self._capacity_factor
       return len(self						) >= int(__lowerCamelCase						)
     def        A__      (						self	:		Tuple						):



       """simple docstring"""
       if len(self._buckets						) <= self._initial_block_size:
         return False
       lowerCAmelCase__										=      len(self._buckets						) * self._capacity_factor / 2
       return len(self						) < limit
     def        A__      (						self	:		List[Any]       ,						__lowerCamelCase	:		int						):



       """simple docstring"""
       lowerCAmelCase__										=      self._buckets
       lowerCAmelCase__										=      [None] * new_size
       lowerCAmelCase__										=      0
       for item in old_buckets:
         if item:
           self._add_item(item.key       ,						item.val						)
     def        A__      (						self	:		Any						):



       """simple docstring"""
       self._resize(len(self._buckets						) * 2						)
     def        A__      (						self	:		Dict						):



       """simple docstring"""
       self._resize(len(self._buckets						) // 2						)
     def        A__      (						self	:		List[str]       ,						__lowerCamelCase	:		KEY						):



       """simple docstring"""
       lowerCAmelCase__										=      self._get_bucket_index(__lowerCamelCase						)
       for _ in range(len(self._buckets						)						):
         yield ind
         lowerCAmelCase__										=      self._get_next_ind(__lowerCamelCase						)
     def        A__      (						self	:		Dict       ,						__lowerCamelCase	:		KEY       ,						__lowerCamelCase	:		VAL						):



       """simple docstring"""
       for ind in self._iterate_buckets(__lowerCamelCase						):
         if self._try_set(__lowerCamelCase       ,						__lowerCamelCase       ,						__lowerCamelCase						):
           break
     def __setitem__(						self	:		str       ,						__lowerCamelCase	:		KEY       ,						__lowerCamelCase	:		VAL						):



       """simple docstring"""
       if self._is_full():
         self._size_up()

       self._add_item(__lowerCamelCase       ,						__lowerCamelCase						)
     def __delitem__(						self	:		Tuple       ,						__lowerCamelCase	:		KEY						):



       """simple docstring"""
       for ind in self._iterate_buckets(__lowerCamelCase						):
         lowerCAmelCase__										=      self._buckets[ind]
         if item is None:
           raise KeyError(__lowerCamelCase						)
         if item is _deleted:
           continue
         if item.key == key:
           lowerCAmelCase__										=      _deleted
           self._len -= 1
           break
       if self._is_sparse():
         self._size_down()
     def __getitem__(						self	:		Optional[int]       ,						__lowerCamelCase	:		KEY						):



       """simple docstring"""
       for ind in self._iterate_buckets(__lowerCamelCase						):
         lowerCAmelCase__										=      self._buckets[ind]
         if item is None:
           break
         if item is _deleted:
           continue
         if item.key == key:
           return item.val
       raise KeyError(__lowerCamelCase						)
     def __len__(						self	:		int						):



       """simple docstring"""
       return self._len
     def __iter__(						self	:		Union[str, Any]						):



       """simple docstring"""
       yield from (item.key for item in self._buckets if item)



     def __repr__(						self	:		Union[str, Any]						):



       """simple docstring"""
       lowerCAmelCase__										=      ''' ,'''.join(
           F"""{item.key}: {item.val}""" for item in self._buckets if item						)
       return F"""HashMap({val_string})"""

